From 87eccd7377d79fc0f61614d24e46a26a978888ca Mon Sep 17 00:00:00 2001 From: Prashant Vasudevan <71649489+vprashrex@users.noreply.github.com> Date: Sat, 24 Oct 2026 12:43:20 +0530 Subject: [PATCH 1/2] feat(observability): one telemetry key per concept via constants Move every span attribute, Sentry tag and metric name into core/telemetry/constants.py. Standard OTel keys (GenAI, HTTP body size, db.system, user.id) come from semconv; tenant ids are always strings. Replaces kaapi.organization_id/project_id with tenant.*, *.job_id with job.id, kaapi.trace_id with correlation_id, llm.provider/llm.request.model with gen_ai.*, and llm.call.total/errors with llm.call.count/error. log_context now uses org_id. Existing key names are otherwise unchanged so dashboards keep working. --- backend/app/api/deps.py | 11 +- backend/app/api/routes/collections.py | 6 +- backend/app/api/routes/guardrails.py | 15 ++- backend/app/api/routes/llm.py | 18 ++- backend/app/celery/celery_app.py | 5 +- backend/app/core/exception_handlers.py | 17 ++- backend/app/core/middleware.py | 21 ++-- backend/app/core/telemetry/constants.py | 111 ++++++++++++++++++ backend/app/core/telemetry/context.py | 9 +- backend/app/core/telemetry/db.py | 58 +++++---- backend/app/core/telemetry/http.py | 28 +++-- backend/app/core/telemetry/llm.py | 89 +++++++++----- backend/app/core/telemetry/metrics.py | 36 ++++-- .../services/collections/create_collection.py | 29 +++-- .../services/collections/delete_collection.py | 21 ++-- backend/app/services/guardrails/jobs.py | 31 +++-- backend/app/services/llm/guardrails.py | 16 ++- backend/app/services/llm/jobs.py | 75 +++++++----- .../tests/core/telemetry/test_telemetry_db.py | 2 +- .../core/telemetry/test_telemetry_llm.py | 12 +- backend/app/tests/core/test_rate_monitor.py | 2 +- backend/app/utils.py | 11 +- .../kaapi-observability-ARCHITECTURE.md | 2 +- docs/wiki/cross-cutting/observability.md | 1 + 24 files changed, 447 insertions(+), 179 deletions(-) create mode 100644 backend/app/core/telemetry/constants.py diff --git a/backend/app/api/deps.py b/backend/app/api/deps.py index a35dc37e1..483092355 100644 --- a/backend/app/api/deps.py +++ b/backend/app/api/deps.py @@ -11,6 +11,11 @@ from sqlmodel import and_, select +from app.core.telemetry.constants import ( + TENANT_ORG_ID, + TENANT_PROJECT_ID, + USER_ID, +) from app.core import security from app.core.config import settings from app.core.db import engine @@ -48,11 +53,11 @@ def _set_tenant_span_attributes(auth_context: AuthContext) -> None: """Stamp org/project on the span, log context and Sentry scope after auth.""" span = trace.get_current_span() if span.is_recording(): - span.set_attribute("user.id", str(auth_context.user.id)) + span.set_attribute(USER_ID, str(auth_context.user.id)) if auth_context.organization: - span.set_attribute("tenant.org_id", str(auth_context.organization.id)) + span.set_attribute(TENANT_ORG_ID, str(auth_context.organization.id)) if auth_context.project: - span.set_attribute("tenant.project_id", str(auth_context.project.id)) + span.set_attribute(TENANT_PROJECT_ID, str(auth_context.project.id)) org_id = auth_context.organization.id if auth_context.organization else None project_id = auth_context.project.id if auth_context.project else None diff --git a/backend/app/api/routes/collections.py b/backend/app/api/routes/collections.py index 6190a76ac..9d491bf81 100644 --- a/backend/app/api/routes/collections.py +++ b/backend/app/api/routes/collections.py @@ -102,7 +102,7 @@ def create_collection( lifecycle="api.collection.create", action="create", project_id=current_user.project_.id, - organization_id=current_user.organization_.id, + org_id=current_user.organization_.id, ): if request.callback_url: validate_callback_url(str(request.callback_url)) @@ -156,7 +156,7 @@ def delete_collection( action="delete", collection_id=str(collection_id), project_id=current_user.project_.id, - organization_id=current_user.organization_.id, + org_id=current_user.organization_.id, ): if request and request.callback_url: validate_callback_url(str(request.callback_url)) @@ -212,7 +212,7 @@ def update_collection( action="update", collection_id=str(collection_id), project_id=current_user.project_.id, - organization_id=current_user.organization_.id, + org_id=current_user.organization_.id, ): collection_crud = CollectionCrud(session, current_user.project_.id) try: diff --git a/backend/app/api/routes/guardrails.py b/backend/app/api/routes/guardrails.py index 4c3968e99..5722d38ea 100644 --- a/backend/app/api/routes/guardrails.py +++ b/backend/app/api/routes/guardrails.py @@ -6,6 +6,11 @@ from fastapi.responses import JSONResponse from opentelemetry import trace +from app.core.telemetry.constants import ( + JOB_ID, + TENANT_ORG_ID, + TENANT_PROJECT_ID, +) from app.api.deps import AuthContextDep, SessionDep from app.api.docs.guardrails_schemas import ( BAN_LIST_CREATE_EXAMPLE, @@ -116,13 +121,13 @@ def apply_guardrails_endpoint( system="guardrails", lifecycle="api.guardrails.apply", project_id=project_id, - organization_id=organization_id, + org_id=organization_id, callback_enabled=request.callback_url is not None, ): span = trace.get_current_span() if span.is_recording(): - span.set_attribute("kaapi.project_id", project_id) - span.set_attribute("kaapi.organization_id", organization_id) + span.set_attribute(TENANT_PROJECT_ID, str(project_id)) + span.set_attribute(TENANT_ORG_ID, str(organization_id)) span.set_attribute( "guardrails.callback_enabled", request.callback_url is not None ) @@ -138,7 +143,7 @@ def apply_guardrails_endpoint( ) if span.is_recording(): - span.set_attribute("guardrails.job_id", str(job.id)) + span.set_attribute(JOB_ID, str(job.id)) message = ( "Guardrails are being applied; the sanitised text will be delivered via callback." @@ -597,7 +602,7 @@ def get_guardrails_job_status( lifecycle="api.guardrails.status", job_id=str(job_id), project_id=project_id, - organization_id=_current_user.organization_.id, + org_id=_current_user.organization_.id, ): job = JobCrud(session=session).get(job_id=job_id, project_id=project_id) if not job or job.job_type != JobType.LLM_GUARDRAILS: diff --git a/backend/app/api/routes/llm.py b/backend/app/api/routes/llm.py index 91ee0d67e..71561b9df 100644 --- a/backend/app/api/routes/llm.py +++ b/backend/app/api/routes/llm.py @@ -7,6 +7,12 @@ from pydantic import TypeAdapter from sqlmodel import Session +from app.core.telemetry.constants import ( + JOB_ID, + LLM_CALLBACK_ENABLED, + TENANT_ORG_ID, + TENANT_PROJECT_ID, +) from app.api.deps import AuthContextDep, SessionDep from app.api.permissions import Permission, require_permission from app.core.cloud.storage import get_cloud_storage @@ -113,14 +119,14 @@ def llm_call( system="llm-call", lifecycle="api.llm.call", project_id=project_id, - organization_id=organization_id, + org_id=organization_id, callback_enabled=request.callback_url is not None, ): span = trace.get_current_span() if span.is_recording(): - span.set_attribute("kaapi.project_id", project_id) - span.set_attribute("kaapi.organization_id", organization_id) - span.set_attribute("llm.callback_enabled", request.callback_url is not None) + span.set_attribute(TENANT_PROJECT_ID, str(project_id)) + span.set_attribute(TENANT_ORG_ID, str(organization_id)) + span.set_attribute(LLM_CALLBACK_ENABLED, request.callback_url is not None) if request.callback_url: validate_callback_url(str(request.callback_url)) @@ -133,7 +139,7 @@ def llm_call( ) if span.is_recording(): - span.set_attribute("llm.job_id", str(job_id)) + span.set_attribute(JOB_ID, str(job_id)) job_crud = JobCrud(session=session) job = job_crud.get(job_id=job_id, project_id=project_id) @@ -180,7 +186,7 @@ def get_llm_call_status( lifecycle="api.llm.call.status", job_id=job_id, project_id=project_id, - organization_id=_current_user.organization_.id, + org_id=_current_user.organization_.id, ): job_crud = JobCrud(session=session) job = job_crud.get(job_id=job_id, project_id=project_id) diff --git a/backend/app/celery/celery_app.py b/backend/app/celery/celery_app.py index c75a9a984..f87ed4dcb 100644 --- a/backend/app/celery/celery_app.py +++ b/backend/app/celery/celery_app.py @@ -11,6 +11,9 @@ ) from kombu import Exchange, Queue +from app.core.telemetry.constants import ( + CELERY_TASK_NAME, +) from app.core.config import settings from app.core.logger import configure_logging @@ -52,7 +55,7 @@ def _tag_sentry_task( **_: object, ) -> None: """Tag the task's Sentry scope with its name and tenant so Issues can be filtered by both.""" - sentry_sdk.set_tag("celery.task_name", task.name) + sentry_sdk.set_tag(CELERY_TASK_NAME, task.name) task_kwargs = kwargs or {} for tag, key in ( ("tenant.project_id", "project_id"), diff --git a/backend/app/core/exception_handlers.py b/backend/app/core/exception_handlers.py index 71a507f56..17e19269c 100644 --- a/backend/app/core/exception_handlers.py +++ b/backend/app/core/exception_handlers.py @@ -16,6 +16,13 @@ HTTP_503_SERVICE_UNAVAILABLE, ) +from app.core.telemetry.constants import ( + DB_CONSTRAINT, + ERROR_CLASS, + HTTP_METHOD, + HTTP_ROUTE, + HTTP_STATUS_CODE, +) from app.utils import APIResponse logger = logging.getLogger(__name__) @@ -136,11 +143,11 @@ def _tag_error_scope(request: Request, status_code: int) -> None: scope = sentry_sdk.get_current_scope() route = getattr(request.scope.get("route"), "path", None) if route: - scope.set_tag("http.route", route) + scope.set_tag(HTTP_ROUTE, route) scope.set_transaction_name(f"{request.method} {route}") - scope.set_tag("http.method", request.method) - scope.set_tag("http.status_code", str(status_code)) - scope.set_tag("error.class", "server" if status_code >= 500 else "client") + scope.set_tag(HTTP_METHOD, request.method) + scope.set_tag(HTTP_STATUS_CODE, str(status_code)) + scope.set_tag(ERROR_CLASS, "server" if status_code >= 500 else "client") def register_exception_handlers(app: FastAPI) -> None: @@ -181,7 +188,7 @@ async def db_error_handler(request: Request, exc: SQLAlchemyError) -> JSONRespon sqlstate, constraint = _db_error_constraint(exc) _tag_error_scope(request, status_code) if constraint and sentry_sdk.get_client().is_active(): - sentry_sdk.set_tag("db.constraint", constraint) + sentry_sdk.set_tag(DB_CONSTRAINT, constraint) log = logger.error if status_code >= 500 else logger.warning log( f"[db_error_handler] Unhandled DB error | path: {request.url.path}, " diff --git a/backend/app/core/middleware.py b/backend/app/core/middleware.py index daeea828c..b02136c3d 100644 --- a/backend/app/core/middleware.py +++ b/backend/app/core/middleware.py @@ -8,6 +8,13 @@ from starlette.middleware.base import RequestResponseEndpoint from starlette.types import ASGIApp, Receive, Scope, Send +from app.core.telemetry.constants import ( + CORRELATION_ID, + HTTP_METHOD, + HTTP_REQUEST_BODY_SIZE, + HTTP_ROUTE, + HTTP_STATUS_CODE, +) from app.core.config import settings from app.core.logger import log_service_name from app.core.telemetry import record_http_request, record_unmatched_request @@ -89,12 +96,12 @@ async def _log_http_request( span = trace.get_current_span() if span.is_recording(): - span.set_attribute("http.request.body.size", request_body_size) + span.set_attribute(HTTP_REQUEST_BODY_SIZE, request_body_size) if sentry_sdk.get_client().is_active(): - sentry_sdk.set_tag("http.method", method) + sentry_sdk.set_tag(HTTP_METHOD, method) if request_id := correlation_id.get(): - sentry_sdk.set_tag("correlation_id", request_id) + sentry_sdk.set_tag(CORRELATION_ID, request_id) try: response = await call_next(request) @@ -103,8 +110,8 @@ async def _log_http_request( status = 500 http_route = _resolve_http_route(request) if sentry_sdk.get_client().is_active(): - sentry_sdk.set_tag("http.route", http_route) - sentry_sdk.set_tag("http.status_code", str(status)) + sentry_sdk.set_tag(HTTP_ROUTE, http_route) + sentry_sdk.set_tag(HTTP_STATUS_CODE, str(status)) if http_route == UNMATCHED_ROUTE: record_unmatched_request(method=method) elif metrics_enabled: @@ -123,8 +130,8 @@ async def _log_http_request( http_route = _resolve_http_route(request) if sentry_sdk.get_client().is_active(): - sentry_sdk.set_tag("http.route", http_route) - sentry_sdk.set_tag("http.status_code", str(status)) + sentry_sdk.set_tag(HTTP_ROUTE, http_route) + sentry_sdk.set_tag(HTTP_STATUS_CODE, str(status)) if http_route == UNMATCHED_ROUTE: record_unmatched_request(method=method) diff --git a/backend/app/core/telemetry/constants.py b/backend/app/core/telemetry/constants.py new file mode 100644 index 000000000..31144a25b --- /dev/null +++ b/backend/app/core/telemetry/constants.py @@ -0,0 +1,111 @@ +"""Telemetry attribute keys and metric names: one name per concept across spans, tags, metrics and logs. + +Tenant ids are always sent as strings so a single Sentry filter matches spans, errors and logs. +""" + +from opentelemetry.semconv._incubating.attributes.gen_ai_attributes import ( # noqa: F401 (re-exported) + GEN_AI_OPERATION_NAME, + GEN_AI_PROVIDER_NAME, + GEN_AI_REQUEST_FREQUENCY_PENALTY, + GEN_AI_REQUEST_MAX_TOKENS, + GEN_AI_REQUEST_MODEL, + GEN_AI_REQUEST_PRESENCE_PENALTY, + GEN_AI_REQUEST_TEMPERATURE, + GEN_AI_REQUEST_TOP_P, + GEN_AI_RESPONSE_MODEL, + GEN_AI_SYSTEM, + GEN_AI_USAGE_INPUT_TOKENS, + GEN_AI_USAGE_OUTPUT_TOKENS, + GEN_AI_USAGE_REASONING_OUTPUT_TOKENS, +) + +from opentelemetry.semconv._incubating.attributes.db_attributes import ( # noqa: F401 (re-exported) + DB_SYSTEM, +) +from opentelemetry.semconv._incubating.attributes.http_attributes import ( # noqa: F401 (re-exported) + HTTP_REQUEST_BODY_SIZE, +) +from opentelemetry.semconv._incubating.attributes.user_attributes import ( # noqa: F401 (re-exported) + USER_ID, +) + +# Sentry AI Insights keys with no OTel semconv equivalent. +GEN_AI_REQUEST_AVAILABLE_TOOLS = "gen_ai.request.available_tools" +GEN_AI_USAGE_TOTAL_TOKENS = "gen_ai.usage.total_tokens" + +TENANT_ORG_ID = "tenant.org_id" +TENANT_PROJECT_ID = "tenant.project_id" + +CORRELATION_ID = "correlation_id" +JOB_ID = "job.id" +CELERY_TASK_ID = "celery.task_id" +CELERY_TASK_NAME = "celery.task_name" + +LLM_CALL_ID = "llm.call_id" +LLM_CHAIN_ID = "llm.chain_id" +LLM_COMPLETION_TYPE = "llm.completion_type" + +HTTP_METHOD = "http.method" +HTTP_ROUTE = "http.route" +HTTP_STATUS_CODE = "http.status_code" +HTTP_ROUTE_CATEGORY = "http.route_category" +ERROR_CLASS = "error.class" +ALERT_TYPE = "alert.type" +SENTRY_OP = "sentry.op" + +LLM_CONFIG_ID = "llm.config.id" +LLM_CONFIG_VERSION = "llm.config.version" +LLM_CONFIG_IS_STORED = "llm.config.is_stored" +LLM_CALLBACK_ENABLED = "llm.callback_enabled" +LLM_PROXY_URL = "llm.proxy.url" + +CALLBACK_URL = "callback.url" +CALLBACK_STATUS = "callback.status" +CALLBACK_REQUEST_BODY_SIZE = "callback.request.body.size" +CALLBACK_RESPONSE_STATUS_CODE = "callback.response.status_code" +CALLBACK_DURATION_MS = "callback.duration_ms" + +COLLECTION_ID = "collection.id" +COLLECTION_PROVIDER = "collection.provider" + +JOB_TABLE = "job.table" +JOB_STATUS = "job.status" +JOB_TYPE = "job.type" +JOB_ACTION_TYPE = "job.action_type" + +# OTel's stable name for SQLSTATE is db.response.status_code; kept as-is so existing dashboards keep working. +DB_SQLSTATE = "db.sqlstate" +DB_ERROR_NAME = "db.error.name" +DB_CONSTRAINT = "db.constraint" +DB_OPERATION = "db.operation" +DB_ROWS_AFFECTED = "db.rows_affected" + +METRIC_HTTP_REQUEST_COUNT = "http.server.request.count" +METRIC_HTTP_REQUEST_DURATION = "http.server.request.duration" +METRIC_HTTP_REQUEST_BODY_SIZE = "http.server.request.body.size" +METRIC_HTTP_REQUEST_ERROR = "http.server.request.error" +METRIC_HTTP_REQUEST_UNMATCHED = "http.server.request.unmatched" +METRIC_LLM_CALL_COUNT = "llm.call.count" +METRIC_LLM_CALL_ERROR = "llm.call.error" +METRIC_LLM_CALL_DURATION = "llm.call.duration" +METRIC_LLM_TOKENS_INPUT = "llm.tokens.input" +METRIC_LLM_TOKENS_OUTPUT = "llm.tokens.output" +METRIC_LLM_TOKENS_TOTAL = "llm.tokens.total" +METRIC_DB_QUERY_FAILED = "db.query.failed" +METRIC_DB_QUERY_SLOW = "db.query.slow" +METRIC_DB_CONNECTION_OPENED = "db.connection.opened" +METRIC_DB_CONNECTION_CLOSED = "db.connection.closed" +METRIC_DB_CONNECTION_INVALIDATED = "db.connection.invalidated" +METRIC_DB_TRANSACTION_COMMIT = "db.transaction.commit" +METRIC_DB_TRANSACTION_ROLLBACK = "db.transaction.rollback" +METRIC_DB_POOL_ACTIVE = "db.pool.active" +METRIC_DB_POOL_IDLE = "db.pool.idle" +METRIC_DB_POOL_TOTAL = "db.pool.total" +METRIC_DB_POOL_OVERFLOW = "db.pool.overflow" +# Unit in the name predates the naming rule; renaming would break existing dashboards. +METRIC_JOB_PENDING_STALE_COUNT = "jobs.pending.stale.count" +METRIC_JOB_PENDING_STALE_BY_DIMENSION_COUNT = "jobs.pending.stale.by_dimension.count" +METRIC_JOB_PENDING_OLDEST_AGE = "jobs.pending.oldest_age_seconds" +METRIC_JOB_PENDING_OLDEST_AGE_BY_DIMENSION = ( + "jobs.pending.oldest_age_seconds.by_dimension" +) diff --git a/backend/app/core/telemetry/context.py b/backend/app/core/telemetry/context.py index c4c64cb13..39003398d 100644 --- a/backend/app/core/telemetry/context.py +++ b/backend/app/core/telemetry/context.py @@ -15,6 +15,11 @@ "kaapi_log_context", default=None ) +from app.core.telemetry.constants import ( + TENANT_ORG_ID, + TENANT_PROJECT_ID, +) + # Sentry skips "_"-prefixed record attributes, so only LOG_MODULE_FIELD reaches Sentry Logs. LOG_MODULE_FIELD = "log_module" LOG_LOCATION_FIELD = "_log_location" @@ -122,9 +127,9 @@ def set_request_log_context( try: if sentry_sdk.get_client().is_active(): if org_id is not None: - sentry_sdk.set_tag("tenant.org_id", str(org_id)) + sentry_sdk.set_tag(TENANT_ORG_ID, str(org_id)) if project_id is not None: - sentry_sdk.set_tag("tenant.project_id", str(project_id)) + sentry_sdk.set_tag(TENANT_PROJECT_ID, str(project_id)) except Exception: logger.debug("[set_request_log_context] Failed to tag Sentry scope") diff --git a/backend/app/core/telemetry/db.py b/backend/app/core/telemetry/db.py index caab90156..806c36683 100644 --- a/backend/app/core/telemetry/db.py +++ b/backend/app/core/telemetry/db.py @@ -11,14 +11,30 @@ from sqlalchemy.engine.interfaces import DBAPIConnection, DBAPICursor, ExecutionContext from sqlalchemy.pool import ConnectionPoolEntry, Pool, PoolProxiedConnection, QueuePool +from app.core.telemetry.constants import ( + DB_ROWS_AFFECTED, + DB_ERROR_NAME, + DB_OPERATION, + DB_SQLSTATE, + DB_SYSTEM, + METRIC_DB_CONNECTION_CLOSED, + METRIC_DB_CONNECTION_INVALIDATED, + METRIC_DB_CONNECTION_OPENED, + METRIC_DB_POOL_ACTIVE, + METRIC_DB_POOL_IDLE, + METRIC_DB_POOL_OVERFLOW, + METRIC_DB_POOL_TOTAL, + METRIC_DB_QUERY_FAILED, + METRIC_DB_QUERY_SLOW, + METRIC_DB_TRANSACTION_COMMIT, + METRIC_DB_TRANSACTION_ROLLBACK, +) from app.core.telemetry.metrics import emit_sentry_metric logger = logging.getLogger(__name__) DB_SLOW_QUERY_MS: int = 500 -DB_ROWS_ATTRIBUTE: str = "db.rows_affected" - _INSTRUMENTED_ATTR = "_kaapi_db_telemetry_instrumented" _STARTED_AT_ATTR = "_kaapi_db_started_at" _OPERATION_ATTR = "_kaapi_db_operation" @@ -34,14 +50,14 @@ } _DB_CONNECTION_EVENT_METRICS: dict[str, str] = { - "opened": "db.connection.opened", - "closed": "db.connection.closed", - "invalidated": "db.connection.invalidated", + "opened": METRIC_DB_CONNECTION_OPENED, + "closed": METRIC_DB_CONNECTION_CLOSED, + "invalidated": METRIC_DB_CONNECTION_INVALIDATED, } _DB_TRANSACTION_METRICS: dict[str, str] = { - "commit": "db.transaction.commit", - "rollback": "db.transaction.rollback", + "commit": METRIC_DB_TRANSACTION_COMMIT, + "rollback": METRIC_DB_TRANSACTION_ROLLBACK, } @@ -54,11 +70,11 @@ def record_db_query_failed( attrs: dict[str, str | int | float] = {} if operation: - attrs["db.operation"] = operation + attrs[DB_OPERATION] = operation if sqlstate: - attrs["db.sqlstate"] = sqlstate + attrs[DB_SQLSTATE] = sqlstate - emit_sentry_metric("count", "db.query.failed", 1, attributes=attrs) + emit_sentry_metric("count", METRIC_DB_QUERY_FAILED, 1, attributes=attrs) def record_db_slow_query(operation: str | None = None) -> None: @@ -66,9 +82,9 @@ def record_db_slow_query(operation: str | None = None) -> None: attrs: dict[str, str | int | float] = {} if operation: - attrs["db.operation"] = operation + attrs[DB_OPERATION] = operation - emit_sentry_metric("count", "db.query.slow", 1, attributes=attrs) + emit_sentry_metric("count", METRIC_DB_QUERY_SLOW, 1, attributes=attrs) def record_db_connection_event(event: str) -> None: @@ -96,13 +112,13 @@ def _tag_db_error(sqlstate: str | None) -> None: try: span = trace.get_current_span() if span.is_recording(): - span.set_attribute("db.sqlstate", sqlstate) + span.set_attribute(DB_SQLSTATE, sqlstate) if sentry_sdk.get_client().is_active(): - sentry_sdk.set_tag("db.system", "postgresql") - sentry_sdk.set_tag("db.sqlstate", sqlstate) + sentry_sdk.set_tag(DB_SYSTEM, "postgresql") + sentry_sdk.set_tag(DB_SQLSTATE, sqlstate) name = NOTABLE_SQLSTATES.get(sqlstate) if name: - sentry_sdk.set_tag("db.error.name", name) + sentry_sdk.set_tag(DB_ERROR_NAME, name) except Exception: logger.debug("[_tag_db_error] Failed to tag DB error | sqlstate: %s", sqlstate) @@ -116,10 +132,10 @@ def record_db_pool_stats( ) -> None: """Emit SQLAlchemy pool stats as Sentry gauges.""" - emit_sentry_metric("gauge", "db.pool.active", active) - emit_sentry_metric("gauge", "db.pool.idle", idle) - emit_sentry_metric("gauge", "db.pool.total", total) - emit_sentry_metric("gauge", "db.pool.overflow", overflow) + emit_sentry_metric("gauge", METRIC_DB_POOL_ACTIVE, active) + emit_sentry_metric("gauge", METRIC_DB_POOL_IDLE, idle) + emit_sentry_metric("gauge", METRIC_DB_POOL_TOTAL, total) + emit_sentry_metric("gauge", METRIC_DB_POOL_OVERFLOW, overflow) def instrument_db_engine(engine: Engine) -> None: @@ -202,7 +218,7 @@ def _record_db_rows( if span is None or rowcount is None or rowcount < 0: return if span.is_recording(): - span.set_attribute(DB_ROWS_ATTRIBUTE, int(rowcount)) + span.set_attribute(DB_ROWS_AFFECTED, int(rowcount)) @event.listens_for(engine, "handle_error") def _handle_error(exception_context: ExceptionContext) -> None: diff --git a/backend/app/core/telemetry/http.py b/backend/app/core/telemetry/http.py index a11b8640c..674efc5a7 100644 --- a/backend/app/core/telemetry/http.py +++ b/backend/app/core/telemetry/http.py @@ -1,5 +1,15 @@ """HTTP request metrics (count, duration, body size, errors) emitted from core/middleware.py.""" +from app.core.telemetry.constants import ( + HTTP_METHOD, + HTTP_ROUTE, + HTTP_STATUS_CODE, + METRIC_HTTP_REQUEST_BODY_SIZE, + METRIC_HTTP_REQUEST_COUNT, + METRIC_HTTP_REQUEST_DURATION, + METRIC_HTTP_REQUEST_ERROR, + METRIC_HTTP_REQUEST_UNMATCHED, +) from app.core.telemetry.metrics import emit_sentry_metric HTTP_ERROR_STATUS_THRESHOLD: int = 400 @@ -16,34 +26,34 @@ def record_http_request( ) -> None: """Emit HTTP traffic, latency, payload-size and error counters to Sentry.""" attrs: dict[str, str | int | float] = { - "http.method": method, - "http.route": http_route, - "http.status_code": str(status), + HTTP_METHOD: method, + HTTP_ROUTE: http_route, + HTTP_STATUS_CODE: str(status), } - emit_sentry_metric("count", "http.server.request.count", 1, attributes=attrs) + emit_sentry_metric("count", METRIC_HTTP_REQUEST_COUNT, 1, attributes=attrs) emit_sentry_metric( "distribution", - "http.server.request.duration", + METRIC_HTTP_REQUEST_DURATION, duration_ms, unit="millisecond", attributes=attrs, ) emit_sentry_metric( "distribution", - "http.server.request.body.size", + METRIC_HTTP_REQUEST_BODY_SIZE, request_body_size, unit="byte", attributes=attrs, ) if status >= HTTP_ERROR_STATUS_THRESHOLD: - emit_sentry_metric("count", "http.server.request.error", 1, attributes=attrs) + emit_sentry_metric("count", METRIC_HTTP_REQUEST_ERROR, 1, attributes=attrs) def record_unmatched_request(*, method: str) -> None: """Count requests that matched no route (scanners/bots) without a route label.""" emit_sentry_metric( "count", - "http.server.request.unmatched", + METRIC_HTTP_REQUEST_UNMATCHED, 1, - attributes={"http.method": method}, + attributes={HTTP_METHOD: method}, ) diff --git a/backend/app/core/telemetry/llm.py b/backend/app/core/telemetry/llm.py index 0ea7ee37b..c4c7c2c2f 100644 --- a/backend/app/core/telemetry/llm.py +++ b/backend/app/core/telemetry/llm.py @@ -10,6 +10,31 @@ suppress_http_instrumentation as otel_suppress_http_instrumentation, ) +from app.core.telemetry.constants import ( + GEN_AI_OPERATION_NAME, + GEN_AI_PROVIDER_NAME, + GEN_AI_REQUEST_AVAILABLE_TOOLS, + GEN_AI_REQUEST_FREQUENCY_PENALTY, + GEN_AI_REQUEST_MAX_TOKENS, + GEN_AI_REQUEST_MODEL, + GEN_AI_REQUEST_PRESENCE_PENALTY, + GEN_AI_REQUEST_TEMPERATURE, + GEN_AI_REQUEST_TOP_P, + GEN_AI_RESPONSE_MODEL, + GEN_AI_SYSTEM, + GEN_AI_USAGE_INPUT_TOKENS, + GEN_AI_USAGE_OUTPUT_TOKENS, + GEN_AI_USAGE_REASONING_OUTPUT_TOKENS, + GEN_AI_USAGE_TOTAL_TOKENS, + METRIC_LLM_CALL_COUNT, + METRIC_LLM_CALL_DURATION, + METRIC_LLM_CALL_ERROR, + METRIC_LLM_TOKENS_INPUT, + METRIC_LLM_TOKENS_OUTPUT, + METRIC_LLM_TOKENS_TOTAL, + TENANT_ORG_ID, + TENANT_PROJECT_ID, +) from app.core.telemetry.metrics import emit_sentry_metric if TYPE_CHECKING: @@ -24,14 +49,14 @@ def _llm_call_attrs( project_id: int | None, ) -> dict[str, str]: attrs: dict[str, str] = { - "gen_ai.system": provider, - "gen_ai.request.model": model, - "gen_ai.operation.name": operation, + GEN_AI_SYSTEM: provider, + GEN_AI_REQUEST_MODEL: model, + GEN_AI_OPERATION_NAME: operation, } if organization_id is not None: - attrs["kaapi.organization_id"] = str(organization_id) + attrs[TENANT_ORG_ID] = str(organization_id) if project_id is not None: - attrs["kaapi.project_id"] = str(project_id) + attrs[TENANT_PROJECT_ID] = str(project_id) return attrs @@ -44,7 +69,7 @@ def record_llm_call_started( ) -> None: """Emit LLM call-start metric to Sentry.""" attrs = _llm_call_attrs(provider, model, operation, organization_id, project_id) - emit_sentry_metric("count", "llm.call.total", 1, attributes=attrs) + emit_sentry_metric("count", METRIC_LLM_CALL_COUNT, 1, attributes=attrs) def record_llm_call_finished( @@ -64,21 +89,25 @@ def record_llm_call_finished( emit_sentry_metric( "distribution", - "llm.call.duration", + METRIC_LLM_CALL_DURATION, duration_ms, unit="millisecond", attributes=attrs, ) if error: - emit_sentry_metric("count", "llm.call.errors", 1, attributes=attrs) + emit_sentry_metric("count", METRIC_LLM_CALL_ERROR, 1, attributes=attrs) if input_tokens is not None: - emit_sentry_metric("count", "llm.tokens.input", input_tokens, attributes=attrs) + emit_sentry_metric( + "count", METRIC_LLM_TOKENS_INPUT, input_tokens, attributes=attrs + ) if output_tokens is not None: emit_sentry_metric( - "count", "llm.tokens.output", output_tokens, attributes=attrs + "count", METRIC_LLM_TOKENS_OUTPUT, output_tokens, attributes=attrs ) if total_tokens is not None: - emit_sentry_metric("count", "llm.tokens.total", total_tokens, attributes=attrs) + emit_sentry_metric( + "count", METRIC_LLM_TOKENS_TOTAL, total_tokens, attributes=attrs + ) def set_gen_ai_request_attributes( @@ -91,26 +120,24 @@ def set_gen_ai_request_attributes( project_id: int | None, params: Mapping[str, object] | None = None, ) -> None: - """Set OTel GenAI request attributes on `span` (semantic-convention keys + kaapi ids).""" - span.set_attribute("gen_ai.system", provider) - span.set_attribute("gen_ai.provider.name", provider) - span.set_attribute("gen_ai.operation.name", operation) + """Set OTel GenAI request attributes on `span` (semantic-convention keys + tenant ids).""" + span.set_attribute(GEN_AI_SYSTEM, provider) + span.set_attribute(GEN_AI_PROVIDER_NAME, provider) + span.set_attribute(GEN_AI_OPERATION_NAME, operation) if model: - span.set_attribute("gen_ai.request.model", model) + span.set_attribute(GEN_AI_REQUEST_MODEL, model) if organization_id is not None: - span.set_attribute("kaapi.organization_id", organization_id) - span.set_attribute("gen_ai.request.organization_id", organization_id) + span.set_attribute(TENANT_ORG_ID, str(organization_id)) if project_id is not None: - span.set_attribute("kaapi.project_id", project_id) - span.set_attribute("gen_ai.request.project_id", project_id) + span.set_attribute(TENANT_PROJECT_ID, str(project_id)) params = params or {} for attr_key, param_key in ( - ("gen_ai.request.temperature", "temperature"), - ("gen_ai.request.max_tokens", "max_tokens"), - ("gen_ai.request.top_p", "top_p"), - ("gen_ai.request.presence_penalty", "presence_penalty"), - ("gen_ai.request.frequency_penalty", "frequency_penalty"), + (GEN_AI_REQUEST_TEMPERATURE, "temperature"), + (GEN_AI_REQUEST_MAX_TOKENS, "max_tokens"), + (GEN_AI_REQUEST_TOP_P, "top_p"), + (GEN_AI_REQUEST_PRESENCE_PENALTY, "presence_penalty"), + (GEN_AI_REQUEST_FREQUENCY_PENALTY, "frequency_penalty"), ): value = params.get(param_key) if isinstance(value, str | bool | int | float): @@ -118,7 +145,7 @@ def set_gen_ai_request_attributes( tools = params.get("tools") if tools is not None: - span.set_attribute("gen_ai.request.available_tools", json.dumps(tools)) + span.set_attribute(GEN_AI_REQUEST_AVAILABLE_TOOLS, json.dumps(tools)) def set_gen_ai_response_attributes( @@ -131,14 +158,14 @@ def set_gen_ai_response_attributes( output_tokens = usage.output_tokens if reasoning_tokens and usage.input_tokens + output_tokens < usage.total_tokens: output_tokens += reasoning_tokens - span.set_attribute("gen_ai.usage.input_tokens", usage.input_tokens) - span.set_attribute("gen_ai.usage.output_tokens", output_tokens) - span.set_attribute("gen_ai.usage.total_tokens", usage.total_tokens) + span.set_attribute(GEN_AI_USAGE_INPUT_TOKENS, usage.input_tokens) + span.set_attribute(GEN_AI_USAGE_OUTPUT_TOKENS, output_tokens) + span.set_attribute(GEN_AI_USAGE_TOTAL_TOKENS, usage.total_tokens) if reasoning_tokens: - span.set_attribute("gen_ai.usage.reasoning.output_tokens", reasoning_tokens) + span.set_attribute(GEN_AI_USAGE_REASONING_OUTPUT_TOKENS, reasoning_tokens) if response.response and response.response.model: - span.set_attribute("gen_ai.response.model", response.response.model) + span.set_attribute(GEN_AI_RESPONSE_MODEL, response.response.model) @contextmanager diff --git a/backend/app/core/telemetry/metrics.py b/backend/app/core/telemetry/metrics.py index f519612e3..19f8da247 100644 --- a/backend/app/core/telemetry/metrics.py +++ b/backend/app/core/telemetry/metrics.py @@ -6,6 +6,20 @@ import sentry_sdk +from app.core.telemetry.constants import ( + ALERT_TYPE, + HTTP_ROUTE_CATEGORY, + JOB_ACTION_TYPE, + JOB_STATUS, + JOB_TABLE, + JOB_TYPE, + METRIC_JOB_PENDING_OLDEST_AGE, + METRIC_JOB_PENDING_OLDEST_AGE_BY_DIMENSION, + METRIC_JOB_PENDING_STALE_BY_DIMENSION_COUNT, + METRIC_JOB_PENDING_STALE_COUNT, + TENANT_PROJECT_ID, +) + logger = logging.getLogger(__name__) MetricType = Literal["count", "gauge", "distribution"] @@ -47,23 +61,23 @@ def record_stale_pending_jobs( summed together with the table-level count in Sentry dashboards. """ attrs: dict[str, str] = { - "job.table": table, - "job.status": status, + JOB_TABLE: table, + JOB_STATUS: status, } if job_type: - attrs["job.type"] = job_type + attrs[JOB_TYPE] = job_type if action_type: - attrs["job.action_type"] = action_type + attrs[JOB_ACTION_TYPE] = action_type count_metric = ( - "jobs.pending.stale.by_dimension.count" + METRIC_JOB_PENDING_STALE_BY_DIMENSION_COUNT if dimensional - else "jobs.pending.stale.count" + else METRIC_JOB_PENDING_STALE_COUNT ) age_metric = ( - "jobs.pending.oldest_age_seconds.by_dimension" + METRIC_JOB_PENDING_OLDEST_AGE_BY_DIMENSION if dimensional - else "jobs.pending.oldest_age_seconds" + else METRIC_JOB_PENDING_OLDEST_AGE ) emit_sentry_metric( @@ -96,9 +110,9 @@ def record_rate_threshold( if not sentry_sdk.get_client().is_active(): return with sentry_sdk.new_scope() as scope: - scope.set_tag("alert.type", "threshold_rate_monitor") - scope.set_tag("tenant.project_id", project_id) - scope.set_tag("route_category", category) + scope.set_tag(ALERT_TYPE, "threshold_rate_monitor") + scope.set_tag(TENANT_PROJECT_ID, str(project_id)) + scope.set_tag(HTTP_ROUTE_CATEGORY, category) scope.set_extra("request_count", request_count) scope.set_extra("threshold", threshold) sentry_sdk.capture_message( diff --git a/backend/app/services/collections/create_collection.py b/backend/app/services/collections/create_collection.py index 66a6a6e61..267167661 100644 --- a/backend/app/services/collections/create_collection.py +++ b/backend/app/services/collections/create_collection.py @@ -10,6 +10,13 @@ from opentelemetry import trace from asgi_correlation_id import correlation_id +from app.core.telemetry.constants import ( + COLLECTION_ID, + COLLECTION_PROVIDER, + JOB_ID, + TENANT_ORG_ID, + TENANT_PROJECT_ID, +) from app.core.cloud import get_cloud_storage from app.core.db import engine from app.core.telemetry import log_context @@ -198,15 +205,15 @@ def execute_setup_job( collection_job_id=job_id, task_id=task_id, project_id=project_id, - organization_id=organization_id, + org_id=organization_id, ), tracer.start_as_current_span("collections.create.execute_setup_job") as span: - span.set_attribute("collection.job_id", str(job_id)) - span.set_attribute("kaapi.project_id", project_id) - span.set_attribute("kaapi.organization_id", organization_id) + span.set_attribute(JOB_ID, str(job_id)) + span.set_attribute(TENANT_PROJECT_ID, str(project_id)) + span.set_attribute(TENANT_ORG_ID, str(organization_id)) try: creation_request = CreationRequest.model_validate(request) - span.set_attribute("collection.provider", str(creation_request.provider)) + span.set_attribute(COLLECTION_PROVIDER, str(creation_request.provider)) job_uuid = UUID(job_id) trace_id = correlation_id.get() or "N/A" @@ -357,16 +364,16 @@ def execute_batch_job( collection_job_id=job_id, task_id=task_id, project_id=project_id, - organization_id=organization_id, + org_id=organization_id, ), tracer.start_as_current_span("collections.create.execute_batch_job") as span: - span.set_attribute("collection.job_id", str(job_id)) - span.set_attribute("kaapi.project_id", project_id) - span.set_attribute("kaapi.organization_id", organization_id) + span.set_attribute(JOB_ID, str(job_id)) + span.set_attribute(TENANT_PROJECT_ID, str(project_id)) + span.set_attribute(TENANT_ORG_ID, str(organization_id)) try: batch_start_time = time.time() creation_request = CreationRequest.model_validate(request) - span.set_attribute("collection.provider", str(creation_request.provider)) + span.set_attribute(COLLECTION_PROVIDER, str(creation_request.provider)) result = Collection( # pyright: ignore[reportCallIssue] knowledge_base_id=vector_store_id, @@ -505,7 +512,7 @@ def execute_batch_job( success_payload = build_success_payload(collection_job, collection) - span.set_attribute("collection.id", str(collection_id)) + span.set_attribute(COLLECTION_ID, str(collection_id)) logger.info( "[create_collection.execute_batch_job] All batches done, collection created: %s | " diff --git a/backend/app/services/collections/delete_collection.py b/backend/app/services/collections/delete_collection.py index 51d4233ff..4001cfd39 100644 --- a/backend/app/services/collections/delete_collection.py +++ b/backend/app/services/collections/delete_collection.py @@ -9,6 +9,13 @@ from sqlmodel import Session from asgi_correlation_id import correlation_id +from app.core.telemetry.constants import ( + COLLECTION_ID, + COLLECTION_PROVIDER, + JOB_ID, + TENANT_ORG_ID, + TENANT_PROJECT_ID, +) from app.core.db import engine from app.crud import CollectionCrud, CollectionJobCrud from app.models import ( @@ -46,7 +53,7 @@ def start_job( collection_job_id=str(collection_job_id), collection_id=str(request.collection_id), project_id=project_id, - organization_id=organization_id, + org_id=organization_id, ): trace_id = correlation_id.get() or "N/A" @@ -194,12 +201,12 @@ def execute_job( collection_id=str(collection_uuid), task_id=task_id, project_id=project_id, - organization_id=organization_id, + org_id=organization_id, ), tracer.start_as_current_span("collections.delete.execute_job") as span: - span.set_attribute("collection.id", str(collection_uuid)) - span.set_attribute("collection.job_id", str(job_uuid)) - span.set_attribute("kaapi.project_id", project_id) - span.set_attribute("kaapi.organization_id", organization_id) + span.set_attribute(COLLECTION_ID, str(collection_uuid)) + span.set_attribute(JOB_ID, str(job_uuid)) + span.set_attribute(TENANT_PROJECT_ID, str(project_id)) + span.set_attribute(TENANT_ORG_ID, str(organization_id)) try: with Session(engine) as session: @@ -216,7 +223,7 @@ def execute_job( collection = CollectionCrud(session, project_id).read_one( collection_uuid ) - span.set_attribute("collection.provider", str(collection.provider)) + span.set_attribute(COLLECTION_PROVIDER, str(collection.provider)) provider = get_llm_provider( session=session, diff --git a/backend/app/services/guardrails/jobs.py b/backend/app/services/guardrails/jobs.py index 1ddf26c11..d623e7f4f 100644 --- a/backend/app/services/guardrails/jobs.py +++ b/backend/app/services/guardrails/jobs.py @@ -7,6 +7,13 @@ from opentelemetry import trace from sqlmodel import Session +from app.core.telemetry.constants import ( + CALLBACK_STATUS, + CALLBACK_URL, + JOB_ID, + TENANT_ORG_ID, + TENANT_PROJECT_ID, +) from app.celery.utils import start_guardrails_job from app.core.db import engine from app.core.telemetry import log_context @@ -47,10 +54,10 @@ def start_job( tag="guardrails", lifecycle="guardrails.start_job", project_id=project_id, - organization_id=organization_id, + org_id=organization_id, ), tracer.start_as_current_span("guardrails.start_job") as span: - span.set_attribute("kaapi.project_id", project_id) - span.set_attribute("kaapi.organization_id", organization_id) + span.set_attribute(TENANT_PROJECT_ID, str(project_id)) + span.set_attribute(TENANT_ORG_ID, str(organization_id)) job_crud = JobCrud(session=db) # Raw text persisted intentionally: this endpoint inspects unsafe content. @@ -60,7 +67,7 @@ def start_job( project_id=project_id, meta={"request": request.model_dump(mode="json")}, ) - span.set_attribute("guardrails.job_id", str(job.id)) + span.set_attribute(JOB_ID, str(job.id)) logger.debug( f"[start_job] Job row created | job_id={job.id}, trace_id={trace_id}" ) @@ -167,8 +174,8 @@ def _send_failure_callback( payload = APIResponse.failure_response(error=error, metadata=metadata).model_dump() webhook_secret = get_webhook_secret(project_id, organization_id) with tracer.start_as_current_span("guardrails.send_callback") as cb_span: - cb_span.set_attribute("callback.url", callback_url) - cb_span.set_attribute("callback.status", "failure") + cb_span.set_attribute(CALLBACK_URL, callback_url) + cb_span.set_attribute(CALLBACK_STATUS, "failure") send_callback( callback_url=callback_url, data=payload, @@ -187,8 +194,8 @@ def _dispatch_success_callback( return webhook_secret = get_webhook_secret(project_id, organization_id) with tracer.start_as_current_span("guardrails.send_callback") as cb_span: - cb_span.set_attribute("callback.url", callback_url) - cb_span.set_attribute("callback.status", "success") + cb_span.set_attribute(CALLBACK_URL, callback_url) + cb_span.set_attribute(CALLBACK_STATUS, "success") send_callback( callback_url=callback_url, data=payload, @@ -305,11 +312,11 @@ def execute_job( lifecycle="guardrails.execute_job", job_id=job_id, project_id=project_id, - organization_id=organization_id, + org_id=organization_id, ), tracer.start_as_current_span("guardrails.execute_job") as span: - span.set_attribute("guardrails.job_id", job_id) - span.set_attribute("kaapi.project_id", project_id) - span.set_attribute("kaapi.organization_id", organization_id) + span.set_attribute(JOB_ID, str(job_id)) + span.set_attribute(TENANT_PROJECT_ID, str(project_id)) + span.set_attribute(TENANT_ORG_ID, str(organization_id)) # 1. Mark job as in-progress. _update_job(job_uuid, status=JobStatus.PROCESSING, task_id=task_id) diff --git a/backend/app/services/llm/guardrails.py b/backend/app/services/llm/guardrails.py index 61002fed5..9e76c6be4 100644 --- a/backend/app/services/llm/guardrails.py +++ b/backend/app/services/llm/guardrails.py @@ -11,6 +11,10 @@ from opentelemetry import trace from tenacity import retry, retry_if_exception, stop_after_attempt, wait_exponential +from app.core.telemetry.constants import ( + TENANT_ORG_ID, + TENANT_PROJECT_ID, +) from app.core.config import settings from app.core.retry import is_transient from app.models.llm.request import InlineValidator, InlineValidatorTypeEnum, Validator @@ -87,8 +91,8 @@ def proxy_guardrails_request( tracer.start_as_current_span( f"guardrails.proxy {method} {path}", attributes={ - "kaapi.organization_id": str(organization_id), - "kaapi.project_id": str(project_id), + TENANT_ORG_ID: str(organization_id), + TENANT_PROJECT_ID: str(project_id), }, ), httpx.Client(timeout=GUARDRAILS_PROXY_TIMEOUT_SECONDS) as client, @@ -372,8 +376,8 @@ def run_guardrails_validation( "guardrails.proxy.validate", attributes={ "kaapi.job_id": str(job_id), - "kaapi.organization_id": str(organization_id), - "kaapi.project_id": str(project_id), + TENANT_ORG_ID: str(organization_id), + TENANT_PROJECT_ID: str(project_id), "kaapi.validator_count": len(validators), }, ), @@ -469,8 +473,8 @@ def _build_params(validator_ids: list[UUID]) -> dict[str, Any]: tracer.start_as_current_span( "guardrails.proxy.list_validator_configs", attributes={ - "kaapi.organization_id": str(organization_id), - "kaapi.project_id": str(project_id), + TENANT_ORG_ID: str(organization_id), + TENANT_PROJECT_ID: str(project_id), }, ), httpx.Client(timeout=10.0) as client, diff --git a/backend/app/services/llm/jobs.py b/backend/app/services/llm/jobs.py index 09c46ce49..edac65bb4 100644 --- a/backend/app/services/llm/jobs.py +++ b/backend/app/services/llm/jobs.py @@ -18,6 +18,25 @@ from sqlmodel import Session from tenacity import retry, retry_if_exception, stop_after_attempt, wait_exponential +from app.core.telemetry.constants import ( + CALLBACK_STATUS, + CALLBACK_URL, + CELERY_TASK_ID, + CORRELATION_ID, + GEN_AI_PROVIDER_NAME, + GEN_AI_REQUEST_MODEL, + JOB_ID, + LLM_CALL_ID, + LLM_CHAIN_ID, + LLM_COMPLETION_TYPE, + LLM_CONFIG_ID, + LLM_CONFIG_IS_STORED, + LLM_CONFIG_VERSION, + LLM_PROXY_URL, + SENTRY_OP, + TENANT_ORG_ID, + TENANT_PROJECT_ID, +) from app.celery.utils import start_llm_chain_job, start_llm_job from app.core.db import engine from app.core.providers import Provider @@ -104,19 +123,19 @@ def _set_traceability_attributes( task_id: str | None = None, ) -> None: if job_id is not None: - span.set_attribute("llm.job_id", str(job_id)) + span.set_attribute(JOB_ID, str(job_id)) if llm_call_id is not None: - span.set_attribute("llm.call_id", str(llm_call_id)) + span.set_attribute(LLM_CALL_ID, str(llm_call_id)) if chain_id is not None: - span.set_attribute("llm.chain_id", str(chain_id)) + span.set_attribute(LLM_CHAIN_ID, str(chain_id)) if trace_id is not None: - span.set_attribute("kaapi.trace_id", trace_id) + span.set_attribute(CORRELATION_ID, trace_id) if project_id is not None: - span.set_attribute("kaapi.project_id", project_id) + span.set_attribute(TENANT_PROJECT_ID, str(project_id)) if organization_id is not None: - span.set_attribute("kaapi.organization_id", organization_id) + span.set_attribute(TENANT_ORG_ID, str(organization_id)) if task_id is not None: - span.set_attribute("celery.task_id", task_id) + span.set_attribute(CELERY_TASK_ID, task_id) def _execute_provider_call( @@ -152,7 +171,7 @@ def start_job( tag="llm-call", lifecycle="llm.call.start_job", project_id=project_id, - organization_id=organization_id, + org_id=organization_id, ), tracer.start_as_current_span("llm.start_job") as span: trace_id = correlation_id.get() or "N/A" _set_traceability_attributes( @@ -220,7 +239,7 @@ def start_chain_job( lifecycle="llm.chain.start_job", job_id=job.id, project_id=project_id, - organization_id=organization_id, + org_id=organization_id, ), tracer.start_as_current_span("llm.chain.start_job") as span: _set_traceability_attributes( span, @@ -297,8 +316,8 @@ def handle_job_error( if callback_url: webhook_secret = get_webhook_secret(project_id, organization_id) with tracer.start_as_current_span("llm.send_callback") as cb_span: - cb_span.set_attribute("callback.url", callback_url) - cb_span.set_attribute("callback.status", "failure") + cb_span.set_attribute(CALLBACK_URL, callback_url) + cb_span.set_attribute(CALLBACK_STATUS, "failure") _set_traceability_attributes( cb_span, job_id=job_id, @@ -651,10 +670,10 @@ def _load_config_blob( ) -> tuple[ConfigBlob | None, str | None]: """Resolve a stored or ad-hoc config, filling in the default audio provider.""" with ctx.span("llm.resolve_config") as cfg_span: - cfg_span.set_attribute("llm.config.is_stored", config.is_stored_config) + cfg_span.set_attribute(LLM_CONFIG_IS_STORED, config.is_stored_config) if config.is_stored_config: - cfg_span.set_attribute("llm.config.id", str(config.id)) - cfg_span.set_attribute("llm.config.version", str(config.version)) + cfg_span.set_attribute(LLM_CONFIG_ID, str(config.id)) + cfg_span.set_attribute(LLM_CONFIG_VERSION, str(config.version)) config_crud = ConfigVersionCrud( session=session, project_id=ctx.project_id, config_id=config.id ) @@ -951,8 +970,8 @@ def _execute_proxy_call( proxy_data: dict | None = None with ctx.span("llm.proxy.execute") as proxy_span: - proxy_span.set_attribute("llm.provider", Provider.PROXY.value) - proxy_span.set_attribute("llm.proxy.url", client_llm_url) + proxy_span.set_attribute(GEN_AI_PROVIDER_NAME, Provider.PROXY.value) + proxy_span.set_attribute(LLM_PROXY_URL, client_llm_url) try: with suppress_http_instrumentation(): @@ -1288,10 +1307,10 @@ def execute_llm_call( with ctx.span("llm.create_call_record") as create_span: create_span.set_attribute( - "llm.provider", str(completion_config.provider) + GEN_AI_PROVIDER_NAME, str(completion_config.provider) ) if model_name: - create_span.set_attribute("llm.request.model", model_name) + create_span.set_attribute(GEN_AI_REQUEST_MODEL, model_name) try: _create_call_record( session, @@ -1342,9 +1361,9 @@ def execute_llm_call( ai_span_name = f"chat {model_name}" if model_name else f"chat {provider_name}" with ctx.span(ai_span_name) as ai_span: - ai_span.set_attribute("sentry.op", "gen_ai.chat") + ai_span.set_attribute(SENTRY_OP, "gen_ai.chat") if completion_type: - ai_span.set_attribute("completion_type", completion_type) + ai_span.set_attribute(LLM_COMPLETION_TYPE, completion_type) set_gen_ai_request_attributes( ai_span, provider=provider_name, @@ -1360,14 +1379,16 @@ def execute_llm_call( resolved_input_context(query.input) as resolved_input, ctx.span("llm.provider.execute") as provider_span, ): - provider_span.set_attribute("llm.provider", provider_name) + provider_span.set_attribute(GEN_AI_PROVIDER_NAME, provider_name) provider_span.set_attribute( "llm.operation.name", "provider.execute" ) if completion_type: - provider_span.set_attribute("completion_type", completion_type) + provider_span.set_attribute( + LLM_COMPLETION_TYPE, completion_type + ) if model_name: - provider_span.set_attribute("llm.request.model", model_name) + provider_span.set_attribute(GEN_AI_REQUEST_MODEL, model_name) # real blocking network call to referenced provider response, error = _execute_provider_call( func=provider_instance.execute, @@ -1473,7 +1494,7 @@ def execute_job( job_id=job_uuid, task_id=task_id, project_id=project_id, - organization_id=organization_id, + org_id=organization_id, ): _set_traceability_attributes( trace.get_current_span(), @@ -1542,8 +1563,8 @@ def execute_job( if callback_url_str: webhook_secret = get_webhook_secret(project_id, organization_id) with tracer.start_as_current_span("llm.send_callback") as cb_span: - cb_span.set_attribute("callback.url", callback_url_str) - cb_span.set_attribute("callback.status", "success") + cb_span.set_attribute(CALLBACK_URL, callback_url_str) + cb_span.set_attribute(CALLBACK_STATUS, "success") _set_traceability_attributes( cb_span, job_id=job_uuid, @@ -1647,7 +1668,7 @@ def execute_chain_job( job_id=job_uuid, task_id=task_id, project_id=project_id, - organization_id=organization_id, + org_id=organization_id, total_blocks=len(request.blocks), ): _set_traceability_attributes( diff --git a/backend/app/tests/core/telemetry/test_telemetry_db.py b/backend/app/tests/core/telemetry/test_telemetry_db.py index 5fa3a75f4..34374b4cc 100644 --- a/backend/app/tests/core/telemetry/test_telemetry_db.py +++ b/backend/app/tests/core/telemetry/test_telemetry_db.py @@ -253,7 +253,7 @@ def _attach_span( rows_calls = [ c for c in span.set_attribute.call_args_list - if c.args[0] == telemetry.DB_ROWS_ATTRIBUTE + if c.args[0] == telemetry.DB_ROWS_AFFECTED ] assert rows_calls assert 1 in [c.args[1] for c in rows_calls] diff --git a/backend/app/tests/core/telemetry/test_telemetry_llm.py b/backend/app/tests/core/telemetry/test_telemetry_llm.py index 20ee05d87..603e8a160 100644 --- a/backend/app/tests/core/telemetry/test_telemetry_llm.py +++ b/backend/app/tests/core/telemetry/test_telemetry_llm.py @@ -26,10 +26,10 @@ def test_started_emits_count_with_tenant_attrs(self) -> None: "openai", "gpt-4o", "chat", organization_id=1, project_id=2 ) - attrs = _metric_calls(fake)["llm.call.total"]["attributes"] + attrs = _metric_calls(fake)["llm.call.count"]["attributes"] assert attrs["gen_ai.system"] == "openai" - assert attrs["kaapi.organization_id"] == "1" - assert attrs["kaapi.project_id"] == "2" + assert attrs["tenant.org_id"] == "1" + assert attrs["tenant.project_id"] == "2" def test_finished_emits_duration_tokens_and_error(self) -> None: fake = _active_sentry() @@ -47,11 +47,11 @@ def test_finished_emits_duration_tokens_and_error(self) -> None: names = _metric_calls(fake) assert names["llm.call.duration"]["value"] == 120.5 - assert names["llm.call.errors"]["value"] == 1 + assert names["llm.call.error"]["value"] == 1 assert names["llm.tokens.input"]["value"] == 10 assert names["llm.tokens.output"]["value"] == 20 assert names["llm.tokens.total"]["value"] == 30 - assert "kaapi.organization_id" not in names["llm.call.duration"]["attributes"] + assert "tenant.org_id" not in names["llm.call.duration"]["attributes"] class TestGenAiSpanAttributes: @@ -68,7 +68,7 @@ def test_request_attributes_include_params_and_tools(self) -> None: ) set_calls = {c.args[0]: c.args[1] for c in span.set_attribute.call_args_list} assert set_calls["gen_ai.request.model"] == "gpt-4o" - assert set_calls["kaapi.organization_id"] == 1 + assert set_calls["tenant.org_id"] == "1" assert set_calls["gen_ai.request.temperature"] == 0.2 assert "gen_ai.request.max_tokens" not in set_calls assert set_calls["gen_ai.request.available_tools"] == '[{"name": "t"}]' diff --git a/backend/app/tests/core/test_rate_monitor.py b/backend/app/tests/core/test_rate_monitor.py index 131ed30cb..254722763 100644 --- a/backend/app/tests/core/test_rate_monitor.py +++ b/backend/app/tests/core/test_rate_monitor.py @@ -179,7 +179,7 @@ def test_emits_warning_message_with_tags(self): capture.assert_called_once() assert capture.call_args.kwargs["level"] == "warning" scope.set_tag.assert_any_call("alert.type", "threshold_rate_monitor") - scope.set_tag.assert_any_call("tenant.project_id", 616) + scope.set_tag.assert_any_call("tenant.project_id", "616") scope.set_extra.assert_any_call("request_count", 16) scope.set_extra.assert_any_call("threshold", 15) diff --git a/backend/app/utils.py b/backend/app/utils.py index c8692ec96..fbb1eae98 100644 --- a/backend/app/utils.py +++ b/backend/app/utils.py @@ -35,6 +35,11 @@ wait_exponential, ) +from app.core.telemetry.constants import ( + CALLBACK_DURATION_MS, + CALLBACK_REQUEST_BODY_SIZE, + CALLBACK_RESPONSE_STATUS_CODE, +) from app.core import security from app.core.audio_utils import AudioRef from app.core.config import settings @@ -595,15 +600,15 @@ def send_callback( log_payload = {k: v for k, v in data.items() if k != CALLBACK_USER_DATA_KEY} raw_body = json.dumps(data, separators=(",", ":")).encode() span = trace.get_current_span() - span.set_attribute("callback.request.body.size", len(raw_body)) + span.set_attribute(CALLBACK_REQUEST_BODY_SIZE, len(raw_body)) started_at = time.perf_counter() try: response = _post_callback(callback_url, raw_body, webhook_secret) span.set_attribute( - "callback.duration_ms", + CALLBACK_DURATION_MS, round((time.perf_counter() - started_at) * 1000, 2), ) - span.set_attribute("callback.response.status_code", response.status_code) + span.set_attribute(CALLBACK_RESPONSE_STATUS_CODE, response.status_code) return True except requests.RequestException as e: diff --git a/docs/architecture/kaapi-observability-ARCHITECTURE.md b/docs/architecture/kaapi-observability-ARCHITECTURE.md index 2849bd7e3..de105a030 100644 --- a/docs/architecture/kaapi-observability-ARCHITECTURE.md +++ b/docs/architecture/kaapi-observability-ARCHITECTURE.md @@ -128,7 +128,7 @@ All metrics go through `emit_sentry_metric()`, which never raises and does nothi |---|---| | **HTTP** | `http.server.request.count`, `.duration`, `.body.size`, `.error`, `.unmatched` | | **DB** | `db.query.failed` (tagged by Postgres SQLSTATE), `db.query.slow` (≥ `DB_SLOW_QUERY_MS`), `db.connection.opened/closed/invalidated`, `db.transaction.commit/rollback` (rollback ratio), `db.pool.active/idle/total/overflow` (pool saturation) | -| **LLM** | `llm.call.total`, `llm.call.duration`, `llm.call.errors`, `llm.tokens.input/output/total` (by provider and model) | +| **LLM** | `llm.call.count`, `llm.call.duration`, `llm.call.error`, `llm.tokens.input/output/total` (by provider and model) | | **Jobs** | `jobs.pending.stale.count`, `jobs.pending.oldest_age_seconds` (table-level and by job type) | Metrics use the Sentry SDK directly because Sentry does not ingest OpenTelemetry metrics. diff --git a/docs/wiki/cross-cutting/observability.md b/docs/wiki/cross-cutting/observability.md index 43b1dd95a..e39249d0b 100644 --- a/docs/wiki/cross-cutting/observability.md +++ b/docs/wiki/cross-cutting/observability.md @@ -23,6 +23,7 @@ OTel-first, Sentry as sole in-process sink (`instrumenter="otel"`). Init at `mai - Runbook (alerts/dashboards/debugging): `features/sentry-utilization/SENTRY-RUNBOOK.md`. ## Telemetry / misc +- Telemetry keys and metric names live in `core/telemetry/constants.py`; never write a key as a string literal. Standard OTel keys (GenAI, `http.request.body.size`, `db.system`, `user.id`) are re-exported from semconv; only Kaapi-specific keys are defined there. Existing key names are kept as-is so dashboards don't break. One name per concept: `tenant.org_id` / `tenant.project_id` (always strings), `job.id`, `correlation_id`, `llm.completion_type`, `http.route_category`; metrics are `..count|error|duration` with the unit in `unit=`. - `core/telemetry/` package, public API re-exported from `__init__.py`: `tracing.py` (OTel tracer provider, span noise filter, FastAPI instrumentation with `exclude_spans` for ASGI send/receive, `OTEL_SEMCONV_STABILITY_OPT_IN=http/dup` default so server spans carry both old and stable `http.*` names, root-span attributes forwarded into Sentry transaction data, root server spans without `http.route` dropped as scanner/bot traffic, flush), `context.py` (log context, tenant tags, user binding), `metrics.py` (Sentry metric emit, stale-job and rate-threshold monitors), `http.py` (HTTP request metrics, called from `core/middleware.py`; unrouted requests emit only `http.server.request.unmatched` with no route label and no access log), `llm.py` (gen_ai span attributes, LLM call metrics), `db.py` (SQLAlchemy engine hooks, DB spans `db.statement` / `db.rows_affected`, pool/slow-query/transaction metrics). - Log records carry `module_name` = `[] .` (modules: knowledgebase, llm-call, llm-chain, config-version, assessment (API path), evals (2.0), fastevals). Module names are `LogModuleEnum`; `core/telemetry/context.py::_LOGGER_MODULES` maps each logging file's exact logger name (`__name__`) to its module, so a new logging file must be added there. A flow that runs another flow's code sets `log_context(module=LogModuleEnum.X)` at its entry (LLM chain jobs, fast-eval Celery tasks). Unmapped loggers (legacy assessment, shared eval helpers outside a fast-eval task) get `.` without the `[]` prefix. - Example, one log line in both sinks: From af4fd7096342e8949b96e26a48e36255050304b6 Mon Sep 17 00:00:00 2001 From: Prashant Vasudevan <71649489+vprashrex@users.noreply.github.com> Date: Sat, 24 Oct 2026 13:31:35 +0530 Subject: [PATCH 2/2] feat(telemetry): standardize telemetry keys across modules for consistency --- backend/app/api/deps.py | 2 - backend/app/api/routes/guardrails.py | 5 +- backend/app/api/routes/llm.py | 4 +- backend/app/celery/celery_app.py | 10 +- backend/app/core/middleware.py | 2 +- backend/app/core/telemetry/constants.py | 93 +++++-------------- backend/app/core/telemetry/context.py | 6 +- backend/app/core/telemetry/db.py | 35 +++---- backend/app/core/telemetry/http.py | 15 +-- backend/app/core/telemetry/llm.py | 22 ++--- backend/app/core/telemetry/metrics.py | 27 ++---- backend/app/core/telemetry/tracing.py | 8 +- .../services/collections/create_collection.py | 12 +-- .../services/collections/delete_collection.py | 10 +- backend/app/services/llm/guardrails.py | 5 +- backend/app/services/llm/jobs.py | 29 ++---- 16 files changed, 92 insertions(+), 193 deletions(-) diff --git a/backend/app/api/deps.py b/backend/app/api/deps.py index 483092355..66cf69b91 100644 --- a/backend/app/api/deps.py +++ b/backend/app/api/deps.py @@ -14,7 +14,6 @@ from app.core.telemetry.constants import ( TENANT_ORG_ID, TENANT_PROJECT_ID, - USER_ID, ) from app.core import security from app.core.config import settings @@ -53,7 +52,6 @@ def _set_tenant_span_attributes(auth_context: AuthContext) -> None: """Stamp org/project on the span, log context and Sentry scope after auth.""" span = trace.get_current_span() if span.is_recording(): - span.set_attribute(USER_ID, str(auth_context.user.id)) if auth_context.organization: span.set_attribute(TENANT_ORG_ID, str(auth_context.organization.id)) if auth_context.project: diff --git a/backend/app/api/routes/guardrails.py b/backend/app/api/routes/guardrails.py index 5722d38ea..bb1dfd481 100644 --- a/backend/app/api/routes/guardrails.py +++ b/backend/app/api/routes/guardrails.py @@ -7,6 +7,7 @@ from opentelemetry import trace from app.core.telemetry.constants import ( + CALLBACK_ENABLED, JOB_ID, TENANT_ORG_ID, TENANT_PROJECT_ID, @@ -128,9 +129,7 @@ def apply_guardrails_endpoint( if span.is_recording(): span.set_attribute(TENANT_PROJECT_ID, str(project_id)) span.set_attribute(TENANT_ORG_ID, str(organization_id)) - span.set_attribute( - "guardrails.callback_enabled", request.callback_url is not None - ) + span.set_attribute(CALLBACK_ENABLED, request.callback_url is not None) if request.callback_url: validate_callback_url(str(request.callback_url)) diff --git a/backend/app/api/routes/llm.py b/backend/app/api/routes/llm.py index 71561b9df..8db4f4096 100644 --- a/backend/app/api/routes/llm.py +++ b/backend/app/api/routes/llm.py @@ -8,8 +8,8 @@ from sqlmodel import Session from app.core.telemetry.constants import ( + CALLBACK_ENABLED, JOB_ID, - LLM_CALLBACK_ENABLED, TENANT_ORG_ID, TENANT_PROJECT_ID, ) @@ -126,7 +126,7 @@ def llm_call( if span.is_recording(): span.set_attribute(TENANT_PROJECT_ID, str(project_id)) span.set_attribute(TENANT_ORG_ID, str(organization_id)) - span.set_attribute(LLM_CALLBACK_ENABLED, request.callback_url is not None) + span.set_attribute(CALLBACK_ENABLED, request.callback_url is not None) if request.callback_url: validate_callback_url(str(request.callback_url)) diff --git a/backend/app/celery/celery_app.py b/backend/app/celery/celery_app.py index f87ed4dcb..c375cce10 100644 --- a/backend/app/celery/celery_app.py +++ b/backend/app/celery/celery_app.py @@ -11,9 +11,6 @@ ) from kombu import Exchange, Queue -from app.core.telemetry.constants import ( - CELERY_TASK_NAME, -) from app.core.config import settings from app.core.logger import configure_logging @@ -40,6 +37,7 @@ def _initialize_worker_observability() -> None: from sentry_sdk.integrations.sqlalchemy import SqlalchemyIntegration from app.core.telemetry import init_sentry + from app.core.telemetry.constants import TENANT_ORG_ID, TENANT_PROJECT_ID init_sentry( integrations=[ @@ -55,11 +53,11 @@ def _tag_sentry_task( **_: object, ) -> None: """Tag the task's Sentry scope with its name and tenant so Issues can be filtered by both.""" - sentry_sdk.set_tag(CELERY_TASK_NAME, task.name) + sentry_sdk.set_tag("celery.task_name", task.name) task_kwargs = kwargs or {} for tag, key in ( - ("tenant.project_id", "project_id"), - ("tenant.org_id", "organization_id"), + (TENANT_PROJECT_ID, "project_id"), + (TENANT_ORG_ID, "organization_id"), ): if task_kwargs.get(key) is not None: sentry_sdk.set_tag(tag, str(task_kwargs[key])) diff --git a/backend/app/core/middleware.py b/backend/app/core/middleware.py index b02136c3d..84a7c4a6e 100644 --- a/backend/app/core/middleware.py +++ b/backend/app/core/middleware.py @@ -9,9 +9,9 @@ from starlette.types import ASGIApp, Receive, Scope, Send from app.core.telemetry.constants import ( + HTTP_REQUEST_BODY_SIZE, CORRELATION_ID, HTTP_METHOD, - HTTP_REQUEST_BODY_SIZE, HTTP_ROUTE, HTTP_STATUS_CODE, ) diff --git a/backend/app/core/telemetry/constants.py b/backend/app/core/telemetry/constants.py index 31144a25b..492847109 100644 --- a/backend/app/core/telemetry/constants.py +++ b/backend/app/core/telemetry/constants.py @@ -1,6 +1,7 @@ -"""Telemetry attribute keys and metric names: one name per concept across spans, tags, metrics and logs. +"""Telemetry keys shared across modules, so every span, Sentry tag and metric uses one name per concept. -Tenant ids are always sent as strings so a single Sentry filter matches spans, errors and logs. +Only keys used in more than one place, or that dashboards and alerts filter on, live here. +A key used in a single file stays as a string at that call site. """ from opentelemetry.semconv._incubating.attributes.gen_ai_attributes import ( # noqa: F401 (re-exported) @@ -18,94 +19,50 @@ GEN_AI_USAGE_OUTPUT_TOKENS, GEN_AI_USAGE_REASONING_OUTPUT_TOKENS, ) - from opentelemetry.semconv._incubating.attributes.db_attributes import ( # noqa: F401 (re-exported) DB_SYSTEM, ) from opentelemetry.semconv._incubating.attributes.http_attributes import ( # noqa: F401 (re-exported) HTTP_REQUEST_BODY_SIZE, ) -from opentelemetry.semconv._incubating.attributes.user_attributes import ( # noqa: F401 (re-exported) - USER_ID, -) - -# Sentry AI Insights keys with no OTel semconv equivalent. -GEN_AI_REQUEST_AVAILABLE_TOOLS = "gen_ai.request.available_tools" -GEN_AI_USAGE_TOTAL_TOKENS = "gen_ai.usage.total_tokens" +# --- Tenant: always sent as strings so one filter matches spans, errors and logs --- TENANT_ORG_ID = "tenant.org_id" TENANT_PROJECT_ID = "tenant.project_id" +# --- Request and job identity --- CORRELATION_ID = "correlation_id" JOB_ID = "job.id" -CELERY_TASK_ID = "celery.task_id" -CELERY_TASK_NAME = "celery.task_name" -LLM_CALL_ID = "llm.call_id" -LLM_CHAIN_ID = "llm.chain_id" -LLM_COMPLETION_TYPE = "llm.completion_type" +# --- GenAI: OTel semconv keys above, plus Sentry AI Insights keys semconv doesn't define --- +GEN_AI_REQUEST_AVAILABLE_TOOLS = "gen_ai.request.available_tools" +GEN_AI_USAGE_TOTAL_TOKENS = "gen_ai.usage.total_tokens" +# --- HTTP --- HTTP_METHOD = "http.method" HTTP_ROUTE = "http.route" HTTP_STATUS_CODE = "http.status_code" -HTTP_ROUTE_CATEGORY = "http.route_category" + +# --- Database: DB_SYSTEM comes from OTel semconv (imported above) --- +DB_SQLSTATE = "db.sqlstate" +DB_ERROR_NAME = "db.error.name" +DB_CONSTRAINT = "db.constraint" +DB_OPERATION = "db.operation" +DB_ROWS_AFFECTED = "db.rows_affected" + +# --- Logs: fields added to every log record --- +# Sentry skips "_"-prefixed record attributes, so only LOG_MODULE_FIELD reaches Sentry Logs. +LOG_MODULE_FIELD = "log_module" +LOG_LOCATION_FIELD = "_log_location" + +# --- Errors and alerts --- ERROR_CLASS = "error.class" ALERT_TYPE = "alert.type" -SENTRY_OP = "sentry.op" - -LLM_CONFIG_ID = "llm.config.id" -LLM_CONFIG_VERSION = "llm.config.version" -LLM_CONFIG_IS_STORED = "llm.config.is_stored" -LLM_CALLBACK_ENABLED = "llm.callback_enabled" -LLM_PROXY_URL = "llm.proxy.url" +# --- Outbound callbacks (webhooks to the client's callback_url) --- CALLBACK_URL = "callback.url" +CALLBACK_ENABLED = "callback.enabled" CALLBACK_STATUS = "callback.status" CALLBACK_REQUEST_BODY_SIZE = "callback.request.body.size" CALLBACK_RESPONSE_STATUS_CODE = "callback.response.status_code" CALLBACK_DURATION_MS = "callback.duration_ms" - -COLLECTION_ID = "collection.id" -COLLECTION_PROVIDER = "collection.provider" - -JOB_TABLE = "job.table" -JOB_STATUS = "job.status" -JOB_TYPE = "job.type" -JOB_ACTION_TYPE = "job.action_type" - -# OTel's stable name for SQLSTATE is db.response.status_code; kept as-is so existing dashboards keep working. -DB_SQLSTATE = "db.sqlstate" -DB_ERROR_NAME = "db.error.name" -DB_CONSTRAINT = "db.constraint" -DB_OPERATION = "db.operation" -DB_ROWS_AFFECTED = "db.rows_affected" - -METRIC_HTTP_REQUEST_COUNT = "http.server.request.count" -METRIC_HTTP_REQUEST_DURATION = "http.server.request.duration" -METRIC_HTTP_REQUEST_BODY_SIZE = "http.server.request.body.size" -METRIC_HTTP_REQUEST_ERROR = "http.server.request.error" -METRIC_HTTP_REQUEST_UNMATCHED = "http.server.request.unmatched" -METRIC_LLM_CALL_COUNT = "llm.call.count" -METRIC_LLM_CALL_ERROR = "llm.call.error" -METRIC_LLM_CALL_DURATION = "llm.call.duration" -METRIC_LLM_TOKENS_INPUT = "llm.tokens.input" -METRIC_LLM_TOKENS_OUTPUT = "llm.tokens.output" -METRIC_LLM_TOKENS_TOTAL = "llm.tokens.total" -METRIC_DB_QUERY_FAILED = "db.query.failed" -METRIC_DB_QUERY_SLOW = "db.query.slow" -METRIC_DB_CONNECTION_OPENED = "db.connection.opened" -METRIC_DB_CONNECTION_CLOSED = "db.connection.closed" -METRIC_DB_CONNECTION_INVALIDATED = "db.connection.invalidated" -METRIC_DB_TRANSACTION_COMMIT = "db.transaction.commit" -METRIC_DB_TRANSACTION_ROLLBACK = "db.transaction.rollback" -METRIC_DB_POOL_ACTIVE = "db.pool.active" -METRIC_DB_POOL_IDLE = "db.pool.idle" -METRIC_DB_POOL_TOTAL = "db.pool.total" -METRIC_DB_POOL_OVERFLOW = "db.pool.overflow" -# Unit in the name predates the naming rule; renaming would break existing dashboards. -METRIC_JOB_PENDING_STALE_COUNT = "jobs.pending.stale.count" -METRIC_JOB_PENDING_STALE_BY_DIMENSION_COUNT = "jobs.pending.stale.by_dimension.count" -METRIC_JOB_PENDING_OLDEST_AGE = "jobs.pending.oldest_age_seconds" -METRIC_JOB_PENDING_OLDEST_AGE_BY_DIMENSION = ( - "jobs.pending.oldest_age_seconds.by_dimension" -) diff --git a/backend/app/core/telemetry/context.py b/backend/app/core/telemetry/context.py index 39003398d..74f687cb7 100644 --- a/backend/app/core/telemetry/context.py +++ b/backend/app/core/telemetry/context.py @@ -16,14 +16,12 @@ ) from app.core.telemetry.constants import ( + LOG_LOCATION_FIELD, + LOG_MODULE_FIELD, TENANT_ORG_ID, TENANT_PROJECT_ID, ) -# Sentry skips "_"-prefixed record attributes, so only LOG_MODULE_FIELD reaches Sentry Logs. -LOG_MODULE_FIELD = "log_module" -LOG_LOCATION_FIELD = "_log_location" - class LogModuleEnum(StrEnum): KNOWLEDGEBASE = "knowledgebase" diff --git a/backend/app/core/telemetry/db.py b/backend/app/core/telemetry/db.py index 806c36683..54e3ed2d9 100644 --- a/backend/app/core/telemetry/db.py +++ b/backend/app/core/telemetry/db.py @@ -12,22 +12,11 @@ from sqlalchemy.pool import ConnectionPoolEntry, Pool, PoolProxiedConnection, QueuePool from app.core.telemetry.constants import ( - DB_ROWS_AFFECTED, DB_ERROR_NAME, DB_OPERATION, + DB_ROWS_AFFECTED, DB_SQLSTATE, DB_SYSTEM, - METRIC_DB_CONNECTION_CLOSED, - METRIC_DB_CONNECTION_INVALIDATED, - METRIC_DB_CONNECTION_OPENED, - METRIC_DB_POOL_ACTIVE, - METRIC_DB_POOL_IDLE, - METRIC_DB_POOL_OVERFLOW, - METRIC_DB_POOL_TOTAL, - METRIC_DB_QUERY_FAILED, - METRIC_DB_QUERY_SLOW, - METRIC_DB_TRANSACTION_COMMIT, - METRIC_DB_TRANSACTION_ROLLBACK, ) from app.core.telemetry.metrics import emit_sentry_metric @@ -50,14 +39,14 @@ } _DB_CONNECTION_EVENT_METRICS: dict[str, str] = { - "opened": METRIC_DB_CONNECTION_OPENED, - "closed": METRIC_DB_CONNECTION_CLOSED, - "invalidated": METRIC_DB_CONNECTION_INVALIDATED, + "opened": "db.connection.opened", + "closed": "db.connection.closed", + "invalidated": "db.connection.invalidated", } _DB_TRANSACTION_METRICS: dict[str, str] = { - "commit": METRIC_DB_TRANSACTION_COMMIT, - "rollback": METRIC_DB_TRANSACTION_ROLLBACK, + "commit": "db.transaction.commit", + "rollback": "db.transaction.rollback", } @@ -74,7 +63,7 @@ def record_db_query_failed( if sqlstate: attrs[DB_SQLSTATE] = sqlstate - emit_sentry_metric("count", METRIC_DB_QUERY_FAILED, 1, attributes=attrs) + emit_sentry_metric("count", "db.query.failed", 1, attributes=attrs) def record_db_slow_query(operation: str | None = None) -> None: @@ -84,7 +73,7 @@ def record_db_slow_query(operation: str | None = None) -> None: if operation: attrs[DB_OPERATION] = operation - emit_sentry_metric("count", METRIC_DB_QUERY_SLOW, 1, attributes=attrs) + emit_sentry_metric("count", "db.query.slow", 1, attributes=attrs) def record_db_connection_event(event: str) -> None: @@ -132,10 +121,10 @@ def record_db_pool_stats( ) -> None: """Emit SQLAlchemy pool stats as Sentry gauges.""" - emit_sentry_metric("gauge", METRIC_DB_POOL_ACTIVE, active) - emit_sentry_metric("gauge", METRIC_DB_POOL_IDLE, idle) - emit_sentry_metric("gauge", METRIC_DB_POOL_TOTAL, total) - emit_sentry_metric("gauge", METRIC_DB_POOL_OVERFLOW, overflow) + emit_sentry_metric("gauge", "db.pool.active", active) + emit_sentry_metric("gauge", "db.pool.idle", idle) + emit_sentry_metric("gauge", "db.pool.total", total) + emit_sentry_metric("gauge", "db.pool.overflow", overflow) def instrument_db_engine(engine: Engine) -> None: diff --git a/backend/app/core/telemetry/http.py b/backend/app/core/telemetry/http.py index 674efc5a7..b98811900 100644 --- a/backend/app/core/telemetry/http.py +++ b/backend/app/core/telemetry/http.py @@ -4,11 +4,6 @@ HTTP_METHOD, HTTP_ROUTE, HTTP_STATUS_CODE, - METRIC_HTTP_REQUEST_BODY_SIZE, - METRIC_HTTP_REQUEST_COUNT, - METRIC_HTTP_REQUEST_DURATION, - METRIC_HTTP_REQUEST_ERROR, - METRIC_HTTP_REQUEST_UNMATCHED, ) from app.core.telemetry.metrics import emit_sentry_metric @@ -30,30 +25,30 @@ def record_http_request( HTTP_ROUTE: http_route, HTTP_STATUS_CODE: str(status), } - emit_sentry_metric("count", METRIC_HTTP_REQUEST_COUNT, 1, attributes=attrs) + emit_sentry_metric("count", "http.server.request.count", 1, attributes=attrs) emit_sentry_metric( "distribution", - METRIC_HTTP_REQUEST_DURATION, + "http.server.request.duration", duration_ms, unit="millisecond", attributes=attrs, ) emit_sentry_metric( "distribution", - METRIC_HTTP_REQUEST_BODY_SIZE, + "http.server.request.body.size", request_body_size, unit="byte", attributes=attrs, ) if status >= HTTP_ERROR_STATUS_THRESHOLD: - emit_sentry_metric("count", METRIC_HTTP_REQUEST_ERROR, 1, attributes=attrs) + emit_sentry_metric("count", "http.server.request.error", 1, attributes=attrs) def record_unmatched_request(*, method: str) -> None: """Count requests that matched no route (scanners/bots) without a route label.""" emit_sentry_metric( "count", - METRIC_HTTP_REQUEST_UNMATCHED, + "http.server.request.unmatched", 1, attributes={HTTP_METHOD: method}, ) diff --git a/backend/app/core/telemetry/llm.py b/backend/app/core/telemetry/llm.py index c4c7c2c2f..1013627b0 100644 --- a/backend/app/core/telemetry/llm.py +++ b/backend/app/core/telemetry/llm.py @@ -26,12 +26,6 @@ GEN_AI_USAGE_OUTPUT_TOKENS, GEN_AI_USAGE_REASONING_OUTPUT_TOKENS, GEN_AI_USAGE_TOTAL_TOKENS, - METRIC_LLM_CALL_COUNT, - METRIC_LLM_CALL_DURATION, - METRIC_LLM_CALL_ERROR, - METRIC_LLM_TOKENS_INPUT, - METRIC_LLM_TOKENS_OUTPUT, - METRIC_LLM_TOKENS_TOTAL, TENANT_ORG_ID, TENANT_PROJECT_ID, ) @@ -69,7 +63,7 @@ def record_llm_call_started( ) -> None: """Emit LLM call-start metric to Sentry.""" attrs = _llm_call_attrs(provider, model, operation, organization_id, project_id) - emit_sentry_metric("count", METRIC_LLM_CALL_COUNT, 1, attributes=attrs) + emit_sentry_metric("count", "llm.call.count", 1, attributes=attrs) def record_llm_call_finished( @@ -89,25 +83,21 @@ def record_llm_call_finished( emit_sentry_metric( "distribution", - METRIC_LLM_CALL_DURATION, + "llm.call.duration", duration_ms, unit="millisecond", attributes=attrs, ) if error: - emit_sentry_metric("count", METRIC_LLM_CALL_ERROR, 1, attributes=attrs) + emit_sentry_metric("count", "llm.call.error", 1, attributes=attrs) if input_tokens is not None: - emit_sentry_metric( - "count", METRIC_LLM_TOKENS_INPUT, input_tokens, attributes=attrs - ) + emit_sentry_metric("count", "llm.tokens.input", input_tokens, attributes=attrs) if output_tokens is not None: emit_sentry_metric( - "count", METRIC_LLM_TOKENS_OUTPUT, output_tokens, attributes=attrs + "count", "llm.tokens.output", output_tokens, attributes=attrs ) if total_tokens is not None: - emit_sentry_metric( - "count", METRIC_LLM_TOKENS_TOTAL, total_tokens, attributes=attrs - ) + emit_sentry_metric("count", "llm.tokens.total", total_tokens, attributes=attrs) def set_gen_ai_request_attributes( diff --git a/backend/app/core/telemetry/metrics.py b/backend/app/core/telemetry/metrics.py index 19f8da247..cfacac22c 100644 --- a/backend/app/core/telemetry/metrics.py +++ b/backend/app/core/telemetry/metrics.py @@ -8,15 +8,6 @@ from app.core.telemetry.constants import ( ALERT_TYPE, - HTTP_ROUTE_CATEGORY, - JOB_ACTION_TYPE, - JOB_STATUS, - JOB_TABLE, - JOB_TYPE, - METRIC_JOB_PENDING_OLDEST_AGE, - METRIC_JOB_PENDING_OLDEST_AGE_BY_DIMENSION, - METRIC_JOB_PENDING_STALE_BY_DIMENSION_COUNT, - METRIC_JOB_PENDING_STALE_COUNT, TENANT_PROJECT_ID, ) @@ -61,23 +52,23 @@ def record_stale_pending_jobs( summed together with the table-level count in Sentry dashboards. """ attrs: dict[str, str] = { - JOB_TABLE: table, - JOB_STATUS: status, + "job.table": table, + "job.status": status, } if job_type: - attrs[JOB_TYPE] = job_type + attrs["job.type"] = job_type if action_type: - attrs[JOB_ACTION_TYPE] = action_type + attrs["job.action_type"] = action_type count_metric = ( - METRIC_JOB_PENDING_STALE_BY_DIMENSION_COUNT + "jobs.pending.stale.by_dimension.count" if dimensional - else METRIC_JOB_PENDING_STALE_COUNT + else "jobs.pending.stale.count" ) age_metric = ( - METRIC_JOB_PENDING_OLDEST_AGE_BY_DIMENSION + "jobs.pending.oldest_age_seconds.by_dimension" if dimensional - else METRIC_JOB_PENDING_OLDEST_AGE + else "jobs.pending.oldest_age_seconds" ) emit_sentry_metric( @@ -112,7 +103,7 @@ def record_rate_threshold( with sentry_sdk.new_scope() as scope: scope.set_tag(ALERT_TYPE, "threshold_rate_monitor") scope.set_tag(TENANT_PROJECT_ID, str(project_id)) - scope.set_tag(HTTP_ROUTE_CATEGORY, category) + scope.set_tag("http.route_category", category) scope.set_extra("request_count", request_count) scope.set_extra("threshold", threshold) sentry_sdk.capture_message( diff --git a/backend/app/core/telemetry/tracing.py b/backend/app/core/telemetry/tracing.py index ae54488ef..d5e9de83d 100644 --- a/backend/app/core/telemetry/tracing.py +++ b/backend/app/core/telemetry/tracing.py @@ -21,6 +21,7 @@ from sentry_sdk.integrations.opentelemetry import SentryPropagator, SentrySpanProcessor from sentry_sdk.tracing import Span as SentrySpan +from app.core.telemetry.constants import HTTP_ROUTE from app.core.config import settings logger = logging.getLogger(__name__) @@ -31,18 +32,13 @@ os.environ.setdefault(OTEL_SEMCONV_OPT_IN_ENV, OTEL_SEMCONV_OPT_IN_DEFAULT) FASTAPI_EXCLUDED_SPANS: list[Literal["receive", "send"]] = ["receive", "send"] -HTTP_ROUTE_ATTRIBUTE = "http.route" def _should_drop_unrouted_server_span( *, is_root: bool, kind: SpanKind, attributes: Mapping[str, object] | None ) -> bool: """True for a root server span that matched no route: scanner/bot traffic.""" - return ( - is_root - and kind == SpanKind.SERVER - and HTTP_ROUTE_ATTRIBUTE not in (attributes or {}) - ) + return is_root and kind == SpanKind.SERVER and HTTP_ROUTE not in (attributes or {}) def _should_drop_bare_http_trace( diff --git a/backend/app/services/collections/create_collection.py b/backend/app/services/collections/create_collection.py index 267167661..7be7be5d9 100644 --- a/backend/app/services/collections/create_collection.py +++ b/backend/app/services/collections/create_collection.py @@ -11,8 +11,6 @@ from asgi_correlation_id import correlation_id from app.core.telemetry.constants import ( - COLLECTION_ID, - COLLECTION_PROVIDER, JOB_ID, TENANT_ORG_ID, TENANT_PROJECT_ID, @@ -202,7 +200,7 @@ def execute_setup_job( tag="collection", lifecycle="collection.create.execute_setup_job", action="create", - collection_job_id=job_id, + job_id=job_id, task_id=task_id, project_id=project_id, org_id=organization_id, @@ -213,7 +211,7 @@ def execute_setup_job( try: creation_request = CreationRequest.model_validate(request) - span.set_attribute(COLLECTION_PROVIDER, str(creation_request.provider)) + span.set_attribute("collection.provider", str(creation_request.provider)) job_uuid = UUID(job_id) trace_id = correlation_id.get() or "N/A" @@ -361,7 +359,7 @@ def execute_batch_job( tag="collection", lifecycle="collection.create.execute_batch_job", action="create", - collection_job_id=job_id, + job_id=job_id, task_id=task_id, project_id=project_id, org_id=organization_id, @@ -373,7 +371,7 @@ def execute_batch_job( try: batch_start_time = time.time() creation_request = CreationRequest.model_validate(request) - span.set_attribute(COLLECTION_PROVIDER, str(creation_request.provider)) + span.set_attribute("collection.provider", str(creation_request.provider)) result = Collection( # pyright: ignore[reportCallIssue] knowledge_base_id=vector_store_id, @@ -512,7 +510,7 @@ def execute_batch_job( success_payload = build_success_payload(collection_job, collection) - span.set_attribute(COLLECTION_ID, str(collection_id)) + span.set_attribute("collection.id", str(collection_id)) logger.info( "[create_collection.execute_batch_job] All batches done, collection created: %s | " diff --git a/backend/app/services/collections/delete_collection.py b/backend/app/services/collections/delete_collection.py index 4001cfd39..d4f116273 100644 --- a/backend/app/services/collections/delete_collection.py +++ b/backend/app/services/collections/delete_collection.py @@ -10,8 +10,6 @@ from asgi_correlation_id import correlation_id from app.core.telemetry.constants import ( - COLLECTION_ID, - COLLECTION_PROVIDER, JOB_ID, TENANT_ORG_ID, TENANT_PROJECT_ID, @@ -50,7 +48,7 @@ def start_job( tag="collection", lifecycle="collection.delete.start_job", action="delete", - collection_job_id=str(collection_job_id), + job_id=str(collection_job_id), collection_id=str(request.collection_id), project_id=project_id, org_id=organization_id, @@ -197,13 +195,13 @@ def execute_job( tag="collection", lifecycle="collection.delete.execute_job", action="delete", - collection_job_id=job_id, + job_id=job_id, collection_id=str(collection_uuid), task_id=task_id, project_id=project_id, org_id=organization_id, ), tracer.start_as_current_span("collections.delete.execute_job") as span: - span.set_attribute(COLLECTION_ID, str(collection_uuid)) + span.set_attribute("collection.id", str(collection_uuid)) span.set_attribute(JOB_ID, str(job_uuid)) span.set_attribute(TENANT_PROJECT_ID, str(project_id)) span.set_attribute(TENANT_ORG_ID, str(organization_id)) @@ -223,7 +221,7 @@ def execute_job( collection = CollectionCrud(session, project_id).read_one( collection_uuid ) - span.set_attribute(COLLECTION_PROVIDER, str(collection.provider)) + span.set_attribute("collection.provider", str(collection.provider)) provider = get_llm_provider( session=session, diff --git a/backend/app/services/llm/guardrails.py b/backend/app/services/llm/guardrails.py index 9e76c6be4..12d90dd42 100644 --- a/backend/app/services/llm/guardrails.py +++ b/backend/app/services/llm/guardrails.py @@ -12,6 +12,7 @@ from tenacity import retry, retry_if_exception, stop_after_attempt, wait_exponential from app.core.telemetry.constants import ( + JOB_ID, TENANT_ORG_ID, TENANT_PROJECT_ID, ) @@ -375,10 +376,10 @@ def run_guardrails_validation( tracer.start_as_current_span( "guardrails.proxy.validate", attributes={ - "kaapi.job_id": str(job_id), + JOB_ID: str(job_id), TENANT_ORG_ID: str(organization_id), TENANT_PROJECT_ID: str(project_id), - "kaapi.validator_count": len(validators), + "guardrails.validator_count": len(validators), }, ), httpx.Client(timeout=45.0) as client, diff --git a/backend/app/services/llm/jobs.py b/backend/app/services/llm/jobs.py index edac65bb4..34f91b86f 100644 --- a/backend/app/services/llm/jobs.py +++ b/backend/app/services/llm/jobs.py @@ -21,19 +21,10 @@ from app.core.telemetry.constants import ( CALLBACK_STATUS, CALLBACK_URL, - CELERY_TASK_ID, CORRELATION_ID, GEN_AI_PROVIDER_NAME, GEN_AI_REQUEST_MODEL, JOB_ID, - LLM_CALL_ID, - LLM_CHAIN_ID, - LLM_COMPLETION_TYPE, - LLM_CONFIG_ID, - LLM_CONFIG_IS_STORED, - LLM_CONFIG_VERSION, - LLM_PROXY_URL, - SENTRY_OP, TENANT_ORG_ID, TENANT_PROJECT_ID, ) @@ -125,9 +116,9 @@ def _set_traceability_attributes( if job_id is not None: span.set_attribute(JOB_ID, str(job_id)) if llm_call_id is not None: - span.set_attribute(LLM_CALL_ID, str(llm_call_id)) + span.set_attribute("llm.call_id", str(llm_call_id)) if chain_id is not None: - span.set_attribute(LLM_CHAIN_ID, str(chain_id)) + span.set_attribute("llm.chain_id", str(chain_id)) if trace_id is not None: span.set_attribute(CORRELATION_ID, trace_id) if project_id is not None: @@ -135,7 +126,7 @@ def _set_traceability_attributes( if organization_id is not None: span.set_attribute(TENANT_ORG_ID, str(organization_id)) if task_id is not None: - span.set_attribute(CELERY_TASK_ID, task_id) + span.set_attribute("celery.task_id", task_id) def _execute_provider_call( @@ -670,10 +661,10 @@ def _load_config_blob( ) -> tuple[ConfigBlob | None, str | None]: """Resolve a stored or ad-hoc config, filling in the default audio provider.""" with ctx.span("llm.resolve_config") as cfg_span: - cfg_span.set_attribute(LLM_CONFIG_IS_STORED, config.is_stored_config) + cfg_span.set_attribute("llm.config.is_stored", config.is_stored_config) if config.is_stored_config: - cfg_span.set_attribute(LLM_CONFIG_ID, str(config.id)) - cfg_span.set_attribute(LLM_CONFIG_VERSION, str(config.version)) + cfg_span.set_attribute("llm.config.id", str(config.id)) + cfg_span.set_attribute("llm.config.version", str(config.version)) config_crud = ConfigVersionCrud( session=session, project_id=ctx.project_id, config_id=config.id ) @@ -971,7 +962,7 @@ def _execute_proxy_call( with ctx.span("llm.proxy.execute") as proxy_span: proxy_span.set_attribute(GEN_AI_PROVIDER_NAME, Provider.PROXY.value) - proxy_span.set_attribute(LLM_PROXY_URL, client_llm_url) + proxy_span.set_attribute("llm.proxy.url", client_llm_url) try: with suppress_http_instrumentation(): @@ -1361,9 +1352,9 @@ def execute_llm_call( ai_span_name = f"chat {model_name}" if model_name else f"chat {provider_name}" with ctx.span(ai_span_name) as ai_span: - ai_span.set_attribute(SENTRY_OP, "gen_ai.chat") + ai_span.set_attribute("sentry.op", "gen_ai.chat") if completion_type: - ai_span.set_attribute(LLM_COMPLETION_TYPE, completion_type) + ai_span.set_attribute("llm.completion_type", completion_type) set_gen_ai_request_attributes( ai_span, provider=provider_name, @@ -1385,7 +1376,7 @@ def execute_llm_call( ) if completion_type: provider_span.set_attribute( - LLM_COMPLETION_TYPE, completion_type + "llm.completion_type", completion_type ) if model_name: provider_span.set_attribute(GEN_AI_REQUEST_MODEL, model_name)