Skip to content
Open
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
1 change: 1 addition & 0 deletions backend/app/api/deps.py
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,7 @@ def _set_tenant_span_attributes(auth_context: AuthContext) -> None:
set_request_log_context(
org_id=auth_context.organization.id if auth_context.organization else None,
project_id=auth_context.project.id if auth_context.project else None,
user_id=auth_context.user.id,
)


Expand Down
24 changes: 20 additions & 4 deletions backend/app/celery/celery_app.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,10 @@

from app.core.config import settings
from app.core.logger import configure_logging
from app.core.sentry_filters import before_send_transaction_filter
from app.core.sentry_filters import (
before_send_error_filter,
before_send_transaction_filter,
)

logger = logging.getLogger(__name__)
_telemetry_initialized = False
Expand All @@ -38,21 +41,34 @@ def _initialize_worker_observability() -> None:
from sentry_sdk.integrations.logging import LoggingIntegration
from sentry_sdk.integrations.sqlalchemy import SqlalchemyIntegration

from app.core.telemetry import (
SENTRY_PROFILE_LIFECYCLE,
SENTRY_PROFILE_SESSION_SAMPLE_RATE,
resolve_sentry_release,
)

sentry_sdk.init(
dsn=str(settings.SENTRY_DSN),
environment=settings.ENVIRONMENT,
release=settings.API_VERSION,
release=resolve_sentry_release(),
instrumenter="otel",
traces_sample_rate=1.0,
traces_sample_rate=settings.SENTRY_TRACES_SAMPLE_RATE,
sample_rate=settings.SENTRY_ERROR_SAMPLE_RATE,
profile_session_sample_rate=SENTRY_PROFILE_SESSION_SAMPLE_RATE,
profile_lifecycle=SENTRY_PROFILE_LIFECYCLE,
send_default_pii=settings.SENTRY_SEND_DEFAULT_PII,
enable_logs=True,
before_send=before_send_error_filter,
before_send_transaction=before_send_transaction_filter,
integrations=[
LoggingIntegration(
level=logging.INFO,
sentry_logs_level=logging.INFO,
),
# propagate_traces=True links an API request to the task it
# enqueues as one trace; poll-loop re-enqueues opt out per-call.
CeleryIntegration(
propagate_traces=False,
propagate_traces=True,
monitor_beat_tasks=False,
),
],
Expand Down
70 changes: 40 additions & 30 deletions backend/app/celery/tasks/job_execution.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
from app.celery.celery_app import celery_app
from app.celery.utils import gevent_timeout
from app.core.config import settings
from app.core.telemetry import suppress_db_instrumentation

if TYPE_CHECKING:
from app.services.notifications.eval_completion import (
Expand All @@ -36,6 +37,9 @@
# app/core/logger.py and app/celery/utils.py).
DEFAULT_TRACE_ID = "N/A"

# Start a fresh trace per poll cycle; otherwise one trace spans every re-enqueue.
SENTRY_NO_PROPAGATE_HEADERS: dict[str, bool] = {"sentry-propagate-traces": False}


def _set_trace(trace_id: str) -> None:
correlation_id.set(trace_id)
Expand Down Expand Up @@ -88,16 +92,19 @@ def run_llm_job(self, project_id: int, job_id: str, trace_id: str, **kwargs):
from app.services.llm.jobs import execute_job

_set_trace(trace_id)
return _run_with_otel_parent(
self,
lambda: execute_job(
project_id=project_id,
job_id=job_id,
task_id=current_task.request.id,
task_instance=self,
**kwargs,
),
)
# DB spans suppressed job-wide so LLM waterfalls stay clean (drops these queries
# from the Sentry Queries page too — accepted trade-off).
with suppress_db_instrumentation():
return _run_with_otel_parent(
self,
lambda: execute_job(
project_id=project_id,
job_id=job_id,
task_id=current_task.request.id,
task_instance=self,
**kwargs,
),
)


@celery_app.task(bind=True, queue="default", priority=9)
Expand All @@ -106,16 +113,17 @@ def run_llm_chain_job(self, project_id: int, job_id: str, trace_id: str, **kwarg
from app.services.llm.jobs import execute_chain_job

_set_trace(trace_id)
return _run_with_otel_parent(
self,
lambda: execute_chain_job(
project_id=project_id,
job_id=job_id,
task_id=current_task.request.id,
task_instance=self,
**kwargs,
),
)
with suppress_db_instrumentation():
return _run_with_otel_parent(
self,
lambda: execute_chain_job(
project_id=project_id,
job_id=job_id,
task_id=current_task.request.id,
task_instance=self,
**kwargs,
),
)


@celery_app.task(bind=True, queue="default", priority=9)
Expand All @@ -124,16 +132,17 @@ def run_response_job(self, project_id: int, job_id: str, trace_id: str, **kwargs
from app.services.response.jobs import execute_job

_set_trace(trace_id)
return _run_with_otel_parent(
self,
lambda: execute_job(
project_id=project_id,
job_id=job_id,
task_id=current_task.request.id,
task_instance=self,
**kwargs,
),
)
with suppress_db_instrumentation():
return _run_with_otel_parent(
self,
lambda: execute_job(
project_id=project_id,
job_id=job_id,
task_id=current_task.request.id,
task_instance=self,
**kwargs,
),
)


@celery_app.task(bind=True, queue="default", priority=9)
Expand Down Expand Up @@ -393,6 +402,7 @@ def run_assessment_api_batch(
"trace_id": trace_id,
},
countdown=POLL_COUNTDOWN_SECONDS,
headers=SENTRY_NO_PROPAGATE_HEADERS,
)
return result

Expand Down
6 changes: 6 additions & 0 deletions backend/app/core/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -154,6 +154,12 @@ def AWS_S3_BUCKET(self) -> str:
BACKEND_SERVICE_NAME: str = "kaapi-backend"
CRON_SERVICE_NAME: str = "kaapi-cron"

# Defaults preserve current behavior; production .env needs no changes.
SENTRY_TRACES_SAMPLE_RATE: float = 1.0
SENTRY_RELEASE: str | None = None
SENTRY_SEND_DEFAULT_PII: bool = False
SENTRY_ERROR_SAMPLE_RATE: float = 1.0

# Threshold Request Rate per minute
THRESHOLD_LLM_CALL_RATE: int = 15
THRESHOLD_COLLECTIONS_RATE: int = 3
Expand Down
3 changes: 3 additions & 0 deletions backend/app/core/exception_handlers.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
import re
from collections import defaultdict

import sentry_sdk
from fastapi import FastAPI, Request, HTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
Expand Down Expand Up @@ -109,6 +110,8 @@ async def http_exception_handler(

@app.exception_handler(Exception)
async def generic_error_handler(request: Request, exc: Exception) -> JSONResponse:
# Capture within the active span so the Issue links to its trace.
sentry_sdk.capture_exception(exc)
return JSONResponse(
status_code=HTTP_500_INTERNAL_SERVER_ERROR,
content=APIResponse.failure_response(
Expand Down
80 changes: 53 additions & 27 deletions backend/app/core/middleware.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,11 @@
}
)

CRON_PATH_PREFIX: str = f"{settings.API_V1_STR}/cron/"

# Excluded from traces only; logs/metrics and spans inside the handler still emit.
TRACE_EXCLUDED_PATH_PREFIXES: frozenset[str] = frozenset({CRON_PATH_PREFIX})


class StripTrailingSlashMiddleware:
"""
Expand Down Expand Up @@ -50,8 +55,33 @@ def _resolve_http_route(request: Request) -> str:
return templated or "unmatched"


def _emit_http_metrics(
*, method: str, http_route: str, status: int, duration_ms: float
) -> None:
"""Emit HTTP traffic/latency/error counters to Sentry. No-op if the SDK is inactive."""
try:
if not sentry_sdk.get_client().is_active():
return
attrs = {
"http.method": method,
"http.route": http_route,
"http.status_code": str(status),
}
sentry_sdk.metrics.count("http.server.request.count", 1, attributes=attrs)
sentry_sdk.metrics.distribution(
"http.server.request.duration",
duration_ms,
unit="millisecond",
attributes=attrs,
)
if status >= 400:
sentry_sdk.metrics.count("http.server.request.error", 1, attributes=attrs)
except Exception:
logger.debug("[_emit_http_metrics] Sentry metric emit failed")


async def http_request_logger(request: Request, call_next) -> Response:
if request.url.path.startswith(f"{settings.API_V1_STR}/cron/"):
if request.url.path.startswith(CRON_PATH_PREFIX):
with log_service_name(settings.CRON_SERVICE_NAME):
return await _log_http_request(request, call_next)

Expand All @@ -62,6 +92,8 @@ async def _log_http_request(request: Request, call_next) -> Response:
start_time = time.time()
method = request.method
raw_path = request.url.path
# Health/utility paths excluded so they don't skew platform traffic metrics.
metrics_enabled = raw_path not in SILENT_LOG_PATHS

span = trace.get_current_span()
if span.is_recording():
Expand All @@ -78,6 +110,7 @@ async def _log_http_request(request: Request, call_next) -> Response:
try:
response = await call_next(request)
except Exception:
duration_ms = (time.time() - start_time) * 1000
status = 500
http_route = _resolve_http_route(request)
if span.is_recording():
Expand All @@ -88,6 +121,13 @@ async def _log_http_request(request: Request, call_next) -> Response:
sentry_sdk.set_tag("http.route", http_route)
sentry_sdk.set_tag("http.status_code", str(status))
sentry_sdk.set_tag("http.response.status_code", str(status))
if metrics_enabled:
_emit_http_metrics(
method=method,
http_route=http_route,
status=status,
duration_ms=duration_ms,
)
logger.exception("Unhandled exception during request")
raise

Expand All @@ -101,32 +141,18 @@ async def _log_http_request(request: Request, call_next) -> Response:
span.set_attribute("http.response.status_code", status)
span.set_attribute("http.request.duration_ms", round(duration_ms, 2))

if raw_path not in SILENT_LOG_PATHS:
logger.info(f"{method} {raw_path} - {status} [{duration_ms:.2f}ms]")

try:
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.response.status_code", str(status))
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.response.status_code", str(status))

attrs = {
"http.method": method,
"http.route": http_route,
"http.status_code": str(status),
}
sentry_sdk.metrics.count("http.server.request.count", 1, attributes=attrs)
sentry_sdk.metrics.distribution(
"http.server.request.duration",
duration_ms,
unit="millisecond",
attributes=attrs,
)
if status >= 400:
sentry_sdk.metrics.count(
"http.server.request.error", 1, attributes=attrs
)
except Exception:
logger.debug("[http_request_logger] Sentry metric emit failed")
if metrics_enabled:
logger.info(f"{method} {raw_path} - {status} [{duration_ms:.2f}ms]")
_emit_http_metrics(
method=method,
http_route=http_route,
status=status,
duration_ms=duration_ms,
)

return response
40 changes: 40 additions & 0 deletions backend/app/core/sentry_filters.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,13 @@
import re
from typing import Any

from app.core.config import settings

# Request headers stripped from error events while PII is off (case-insensitive).
_SENSITIVE_HEADERS: frozenset[str] = frozenset(
{"authorization", "cookie", "set-cookie", "x-api-key"}
)
_SCRUBBED = "[scrubbed]"

_SQL_OR_CONNECT = re.compile(r"^(select|insert|update|delete|connect)\b", re.IGNORECASE)
_HTTP_SEND_RECEIVE = re.compile(r"http (send|receive)$", re.IGNORECASE)
Expand Down Expand Up @@ -85,3 +92,36 @@ def before_send_transaction_filter(

event["spans"] = filtered
return event


def _scrub_request_pii(event: dict[str, Any]) -> None:
request = event.get("request")
if not isinstance(request, dict):
return

headers = request.get("headers")
if isinstance(headers, dict):
for key in list(headers):
if str(key).lower() in _SENSITIVE_HEADERS:
headers[key] = _SCRUBBED

if "cookies" in request:
request["cookies"] = _SCRUBBED
if request.get("query_string"):
request["query_string"] = _SCRUBBED
if "data" in request:
request["data"] = _SCRUBBED


def before_send_error_filter(
event: dict[str, Any], hint: dict[str, Any]
) -> dict[str, Any] | None:
"""Drop probe/scanner error events; scrub request PII while PII is off."""
try:
if _should_drop_transaction(event):
return None
if not settings.SENTRY_SEND_DEFAULT_PII:
_scrub_request_pii(event)
except Exception:
return event
return event
Loading
Loading