Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 6 additions & 3 deletions backend/app/api/deps.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,10 @@

from sqlmodel import and_, select

from app.core.telemetry.constants import (
TENANT_ORG_ID,
TENANT_PROJECT_ID,
)
from app.core import security
from app.core.config import settings
from app.core.db import engine
Expand Down Expand Up @@ -48,11 +52,10 @@ 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))
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
Expand Down
6 changes: 3 additions & 3 deletions backend/app/api/routes/collections.py
Original file line number Diff line number Diff line change
Expand Up @@ -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))
Expand Down Expand Up @@ -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))
Expand Down Expand Up @@ -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:
Expand Down
20 changes: 12 additions & 8 deletions backend/app/api/routes/guardrails.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,12 @@
from fastapi.responses import JSONResponse
from opentelemetry import trace

from app.core.telemetry.constants import (
CALLBACK_ENABLED,
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,
Expand Down Expand Up @@ -116,16 +122,14 @@ 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(
"guardrails.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(CALLBACK_ENABLED, request.callback_url is not None)

if request.callback_url:
validate_callback_url(str(request.callback_url))
Expand All @@ -138,7 +142,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."
Expand Down Expand Up @@ -597,7 +601,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:
Expand Down
18 changes: 12 additions & 6 deletions backend/app/api/routes/llm.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,12 @@
from pydantic import TypeAdapter
from sqlmodel import Session

from app.core.telemetry.constants import (
CALLBACK_ENABLED,
JOB_ID,
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
Expand Down Expand Up @@ -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(CALLBACK_ENABLED, request.callback_url is not None)

if request.callback_url:
validate_callback_url(str(request.callback_url))
Expand All @@ -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)
Expand Down Expand Up @@ -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)
Expand Down
5 changes: 3 additions & 2 deletions backend/app/celery/celery_app.py
Original file line number Diff line number Diff line change
Expand Up @@ -37,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=[
Expand All @@ -55,8 +56,8 @@ def _tag_sentry_task(
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]))
Expand Down
17 changes: 12 additions & 5 deletions backend/app/core/exception_handlers.py
Original file line number Diff line number Diff line change
Expand Up @@ -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__)
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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}, "
Expand Down
21 changes: 14 additions & 7 deletions backend/app/core/middleware.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,13 @@
from starlette.middleware.base import RequestResponseEndpoint
from starlette.types import ASGIApp, Receive, Scope, Send

from app.core.telemetry.constants import (
HTTP_REQUEST_BODY_SIZE,
CORRELATION_ID,
HTTP_METHOD,
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
Expand Down Expand Up @@ -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)
Expand All @@ -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:
Expand All @@ -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)
Expand Down
68 changes: 68 additions & 0 deletions backend/app/core/telemetry/constants.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,68 @@
"""Telemetry keys shared across modules, so every span, Sentry tag and metric uses one name per concept.

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)
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,
)

# --- 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"

# --- 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"

# --- 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"

# --- 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"
13 changes: 8 additions & 5 deletions backend/app/core/telemetry/context.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,9 +15,12 @@
"kaapi_log_context", default=None
)

# Sentry skips "_"-prefixed record attributes, so only LOG_MODULE_FIELD reaches Sentry Logs.
LOG_MODULE_FIELD = "log_module"
LOG_LOCATION_FIELD = "_log_location"
from app.core.telemetry.constants import (
LOG_LOCATION_FIELD,
LOG_MODULE_FIELD,
TENANT_ORG_ID,
TENANT_PROJECT_ID,
)


class LogModuleEnum(StrEnum):
Expand Down Expand Up @@ -122,9 +125,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")

Expand Down
Loading
Loading