diff --git a/openapi-spec/openapi.json b/openapi-spec/openapi.json index 3235bf0c6..30a400530 100644 --- a/openapi-spec/openapi.json +++ b/openapi-spec/openapi.json @@ -8581,7 +8581,7 @@ "System" ], "summary": "Get Schema Status", - "description": "What the last schema bootstrap pass did, object by object.\n\nThis is the operator's record of the schema apply, replacing the completed\nArgoCD Job the engine took over from. ``state`` is what readiness follows:\nconverged or observed is ready, failed leaves the pod up and NotReady with\nthe cause here.", + "description": "What the last schema bootstrap pass did, object by object.\n\nThis is the operator's record of the schema apply, replacing the completed\nArgoCD Job the engine took over from. ``state`` is what readiness follows:\nconverged or observed is ready, failed leaves the pod up and NotReady with\nthe stage that failed here and the cause in the engine log.", "operationId": "get_schema_status_api_v1_system_schema_get", "responses": { "200": { @@ -35628,7 +35628,7 @@ "error": { "type": "string", "title": "Error", - "description": "Why the pass failed; empty when it did not." + "description": "The stage that failed, its cause in the engine log; empty when none did." }, "counts": { "additionalProperties": { @@ -39342,7 +39342,7 @@ "error" ], "title": "TopicFailure", - "description": "One topic the broker would not do the thing to, and what it said." + "description": "One topic the broker would not do the thing to; the broker's own text is in the log." }, "TopicRemoveResponse": { "properties": { diff --git a/src/dfe_engine/api/errors.py b/src/dfe_engine/api/errors.py index 96c755de8..489c5fadc 100644 --- a/src/dfe_engine/api/errors.py +++ b/src/dfe_engine/api/errors.py @@ -126,6 +126,71 @@ class ErrorCode: UNRESOLVED_REFERENCE = "unresolved_reference" +SERVICE_UNAVAILABLE_MESSAGE = "Backing service is unreachable, retry shortly" +GITOPS_UNAVAILABLE_MESSAGE = "The deploy repo is unreachable, retry after the Retry-After interval" + + +def hide_backend_text(message: str, exc: BaseException, *, event: str, **fields: Any) -> str: + """Log ``exc``'s text under ``event`` and return ``message`` for the caller in its place. + + A backend's own error text can carry statement fragments, user names and password + hashes, so it reaches the engine log and never a response. + + Args: + message: What the caller is told instead. + exc: The backend failure whose text is logged. + event: The log line's message. + **fields: Further structured fields for the log line. + + Returns: + ``message``, unchanged. + """ + logger.warning(event, error=str(exc), **fields) + return message + + +def backend_failure( + status_code: int, code: str, message: str, exc: BaseException, *, event: str, **fields: Any +) -> HTTPException: + """The HTTPException a route raises for a backend failure, its text logged instead of sent. + + Args: + status_code: The HTTP status to answer. + code: The machine-readable error code. + message: What the caller is told. + exc: The backend failure whose text is logged. + event: The log line's message. + **fields: Further structured fields for the log line. + + Returns: + The exception to raise, chained from ``exc`` by the caller. + """ + return HTTPException( + status_code=status_code, + detail={"code": code, "message": hide_backend_text(message, exc, event=event, **fields)}, + ) + + +def engine_message(exc: BaseException, fallback: str, *, event: str, **fields: Any) -> str: + """``exc``'s own message when the engine wrote it, else ``fallback`` with its text logged. + + The error types this serves are raised ``from`` a backend exception whenever they + carry that backend's text, and with no cause when the engine composed the message. + + Args: + exc: The error a route is about to report. + fallback: What the caller is told when ``exc`` carries a backend's text. + event: The log line's message. + **fields: Further structured fields for the log line. + + Returns: + The message that is safe to send. + """ + if exc.__cause__ is None: + return str(exc) + return hide_backend_text(fallback, exc, event=event, **fields) + + def raise_exchange_http(exc: Exception) -> NoReturn: """Map an import/export failure to its HTTP answer. @@ -243,9 +308,18 @@ async def protected_account_handler(request: Request, exc: ProtectedAccountError from scalo.resilience import ServiceUnavailable @app.exception_handler(ServiceUnavailable) - async def service_unavailable_handler(_request: Request, exc: ServiceUnavailable): + async def service_unavailable_handler(request: Request, exc: ServiceUnavailable): waking = bool(getattr(exc, "waking", False)) - message = "Backing service is warming up, retry shortly" if waking else str(exc) + if waking: + message = "Backing service is warming up, retry shortly" + else: + # scalo's message ends with the backend's last error, verbatim. + message = hide_backend_text( + SERVICE_UNAVAILABLE_MESSAGE, + exc, + event="backing service unavailable", + path=request.url.path, + ) body = ErrorResponse( code=ErrorCode.SERVICE_UNAVAILABLE, message=message, @@ -258,10 +332,15 @@ async def service_unavailable_handler(_request: Request, exc: ServiceUnavailable from dfe_engine.gitops.repo import GitopsUnavailableError @app.exception_handler(GitopsUnavailableError) - async def gitops_unavailable_handler(_request: Request, exc: GitopsUnavailableError): + async def gitops_unavailable_handler(request: Request, exc: GitopsUnavailableError): body = ErrorResponse( code=ErrorCode.SERVICE_UNAVAILABLE, - message=str(exc), + message=hide_backend_text( + GITOPS_UNAVAILABLE_MESSAGE, + exc, + event="deploy repo unavailable", + path=request.url.path, + ), context={"service": "gitops", "remote": exc.remote}, ) return JSONResponse( diff --git a/src/dfe_engine/api/task_manager.py b/src/dfe_engine/api/task_manager.py index a76f8c0a4..8cfd8d7dc 100644 --- a/src/dfe_engine/api/task_manager.py +++ b/src/dfe_engine/api/task_manager.py @@ -34,6 +34,9 @@ from typing import Any from pydantic import BaseModel, Field +from scalo.logger import logger + +TASK_FAILED_MESSAGE = "The task failed; the engine log has the reason" def _json_safe(value: Any) -> Any: @@ -159,7 +162,13 @@ def __init__(self, max_completed: int = 1000) -> None: self._max_completed = max_completed def submit( - self, kind: str, coro_fn, *args, held_to: Iterable[str] | None = None, **kwargs + self, + kind: str, + coro_fn, + *args, + held_to: Iterable[str] | None = None, + reportable: tuple[type[Exception], ...] = (), + **kwargs, ) -> TaskInfo: """Submit an async callable for background execution. @@ -167,12 +176,19 @@ def submit( ``_Task`` instance whose ``set_progress()`` method can be called to report progress. + Every failure is logged. Its message reaches the task's readers only when + it is one of ``reportable``; any other failure reads as + :data:`TASK_FAILED_MESSAGE`, because a backend's own error text can carry + statement fragments, user names and password hashes. + Args: kind: The task kind, for ``list(kind=...)``. coro_fn: The coroutine function to run, called with ``*args`` and ``**kwargs``. *args: Positional arguments for ``coro_fn``. held_to: The orgs the task's result is held to. None means it may hold any org's data, so only a reader of every org sees it. + reportable: Exception types whose message the engine composes about the + request, and which a reader may therefore see. **kwargs: Keyword arguments for ``coro_fn``. Returns: @@ -196,9 +212,10 @@ async def _run() -> None: task.status = TaskStatus.CANCELLED task.message = "Cancelled" except Exception as exc: + logger.warning("background task failed", task_id=task.id, kind=kind, error=str(exc)) task.status = TaskStatus.FAILED - task.error = str(exc) - task.message = f"Failed: {exc}" + task.error = str(exc) if isinstance(exc, reportable) else TASK_FAILED_MESSAGE + task.message = f"Failed: {task.error}" finally: task.completed_at = datetime.now(UTC).isoformat() task._progress_event.set() diff --git a/src/dfe_engine/api/v1/apps.py b/src/dfe_engine/api/v1/apps.py index 3c04cabdd..be911c659 100644 --- a/src/dfe_engine/api/v1/apps.py +++ b/src/dfe_engine/api/v1/apps.py @@ -65,7 +65,7 @@ get_source_registry, require_action, ) -from dfe_engine.api.errors import ErrorResponse +from dfe_engine.api.errors import ErrorResponse, backend_failure from dfe_engine.api.v1.app_contracts import read_contract from dfe_engine.api.v1.helm import conflict_error from dfe_engine.api.v1.sampler import check_sample_scope, sample_org_scope @@ -2149,6 +2149,17 @@ def _reader(request: Request, client: Any) -> OperationalReader: return OperationalReader(client, settings.clickhouse.effective_data_database) +def metrics_unavailable(exc: MetricsUnavailableError) -> HTTPException: + """The 503 for an otel read ClickHouse refused, its text logged rather than sent.""" + return backend_failure( + 503, + "metrics_unavailable", + "the telemetry tables could not be read; the engine log has ClickHouse's reason", + exc, + event="otel query failed", + ) + + @router.get("/{service}/{instance}/status") def get_status( service: str, instance: str, user: CurrentUser, request: Request, client: ClickHouseClient @@ -2159,9 +2170,7 @@ def get_status( try: status = _reader(request, client).status(app.telemetry_name) except MetricsUnavailableError as exc: - raise HTTPException( - 503, detail={"code": "metrics_unavailable", "message": str(exc)} - ) from exc + raise metrics_unavailable(exc) from exc return StatusResponse( telemetry_name=status.telemetry_name, reporting=status.reporting, @@ -2181,9 +2190,7 @@ def get_metrics( try: found = _reader(request, client).metrics(app.telemetry_name) except MetricsUnavailableError as exc: - raise HTTPException( - 503, detail={"code": "metrics_unavailable", "message": str(exc)} - ) from exc + raise metrics_unavailable(exc) from exc return MetricsResponse( telemetry_name=found.telemetry_name, window_seconds=found.window_seconds, @@ -2217,9 +2224,7 @@ def get_resource_series( app.telemetry_name, window_seconds=window, bucket_seconds=bucket ) except MetricsUnavailableError as exc: - raise HTTPException( - 503, detail={"code": "metrics_unavailable", "message": str(exc)} - ) from exc + raise metrics_unavailable(exc) from exc return ResourceSeriesResponse( telemetry_name=app.telemetry_name, window_seconds=window, diff --git a/src/dfe_engine/api/v1/discovery.py b/src/dfe_engine/api/v1/discovery.py index ab0416602..e1b909687 100644 --- a/src/dfe_engine/api/v1/discovery.py +++ b/src/dfe_engine/api/v1/discovery.py @@ -21,10 +21,17 @@ from pydantic import BaseModel from dfe_engine.api.deps import CurrentUser, require_action +from dfe_engine.api.errors import backend_failure from dfe_engine.auth.rbac_scopes import scopes_dict router = APIRouter(prefix="/discovery", tags=["discovery"]) +_QUERY_FAILED = "The ClickHouse metadata query failed; the engine log has ClickHouse's reason" + + +def _query_failed(exc: Exception) -> HTTPException: + return backend_failure(500, "query_error", _QUERY_FAILED, exc, event="discovery query failed") + # -- Response models ----------------------------------------- @@ -75,13 +82,13 @@ def _get_ch_client(request: Request): # Discovery is an admin-level cross-database listing, so it reads over "default". return conn_registry.get_client("default") except Exception as exc: - raise HTTPException( - status_code=503, - detail={ - "code": "connection_error", - "message": f"Cannot connect to ClickHouse: {exc}", - }, - ) + raise backend_failure( + 503, + "connection_error", + "Cannot connect to ClickHouse; the engine log has the reason", + exc, + event="discovery: ClickHouse connection failed", + ) from exc # -- Endpoints ----------------------------------------------- @@ -107,7 +114,7 @@ async def list_databases( if row[0] not in ("system", "information_schema", "INFORMATION_SCHEMA") ] except Exception as exc: - raise HTTPException(status_code=500, detail={"code": "query_error", "message": str(exc)}) + raise _query_failed(exc) from exc @router.get( @@ -149,7 +156,7 @@ async def list_tables( for row in result.result_rows ] except Exception as exc: - raise HTTPException(status_code=500, detail={"code": "query_error", "message": str(exc)}) + raise _query_failed(exc) from exc @router.get( @@ -195,4 +202,4 @@ async def list_columns( except HTTPException: raise except Exception as exc: - raise HTTPException(status_code=500, detail={"code": "query_error", "message": str(exc)}) + raise _query_failed(exc) from exc diff --git a/src/dfe_engine/api/v1/governance.py b/src/dfe_engine/api/v1/governance.py index 08c9b40f5..04bb0e2a2 100644 --- a/src/dfe_engine/api/v1/governance.py +++ b/src/dfe_engine/api/v1/governance.py @@ -24,9 +24,9 @@ from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response from pydantic import BaseModel -from scalo.logger import logger from dfe_engine.api.deps import CurrentUser, live_role_config, require_action +from dfe_engine.api.errors import backend_failure from dfe_engine.api.review import apply_review_headers from dfe_engine.api.write_turn import WRITE_TURN from dfe_engine.appmgmt import contract @@ -446,10 +446,12 @@ def reconcile_ch_rbac_endpoint(user: CurrentUser, request: Request) -> dict[str, try: admin_client = ch_admin_client(settings) except Exception as exc: - logger.warning("CH RBAC reconcile: ClickHouse unreachable", error=str(exc)) - raise HTTPException( - status_code=503, - detail={"code": "clickhouse_unavailable", "message": _CH_UNAVAILABLE}, + raise backend_failure( + 503, + "clickhouse_unavailable", + _CH_UNAVAILABLE, + exc, + event="CH RBAC reconcile: ClickHouse unreachable", ) from exc try: @@ -463,10 +465,12 @@ def reconcile_ch_rbac_endpoint(user: CurrentUser, request: Request) -> dict[str, except Exception as exc: if not is_connection_error(exc): raise - logger.warning("CH RBAC reconcile: ClickHouse connection lost", error=str(exc)) - raise HTTPException( - status_code=503, - detail={"code": "clickhouse_unavailable", "message": _CH_UNAVAILABLE}, + raise backend_failure( + 503, + "clickhouse_unavailable", + _CH_UNAVAILABLE, + exc, + event="CH RBAC reconcile: ClickHouse connection lost", ) from exc # A full run covers every identity, so a startup retry still waiting stands down. note_ch_rbac_reconciled(request.app.state) diff --git a/src/dfe_engine/api/v1/kafka_topics.py b/src/dfe_engine/api/v1/kafka_topics.py index 53bdebb80..8052cb186 100644 --- a/src/dfe_engine/api/v1/kafka_topics.py +++ b/src/dfe_engine/api/v1/kafka_topics.py @@ -51,7 +51,7 @@ class TopicFailure(BaseModel): - """One topic the broker would not do the thing to, and what it said.""" + """One topic the broker would not do the thing to; the broker's own text is in the log.""" name: str error: str diff --git a/src/dfe_engine/api/v1/oidc_login.py b/src/dfe_engine/api/v1/oidc_login.py index c6b058b53..11bfbfcfb 100644 --- a/src/dfe_engine/api/v1/oidc_login.py +++ b/src/dfe_engine/api/v1/oidc_login.py @@ -239,10 +239,14 @@ async def oidc_callback( logger.warning("OIDC callback failed", provider=provider, error=str(exc)) # A refused credential is an audit event, not just an operational log line. audit_login_denied("unknown", "oidc", _get_client_ip(request), str(exc)) + # The IdP's own answer stays in the log: this route takes unauthenticated callers. raise HTTPException( status_code=401, - detail={"code": "unauthorized", "message": f"OIDC login failed: {exc}"}, - ) + detail={ + "code": "unauthorized", + "message": "OIDC login failed; the engine log has the reason", + }, + ) from exc if not identity.subject: audit_login_denied("unknown", "oidc", _get_client_ip(request), "no_subject") diff --git a/src/dfe_engine/api/v1/orgs.py b/src/dfe_engine/api/v1/orgs.py index 12eaddafe..10f940cf0 100644 --- a/src/dfe_engine/api/v1/orgs.py +++ b/src/dfe_engine/api/v1/orgs.py @@ -34,6 +34,7 @@ is_action_allowed, require_action, ) +from dfe_engine.api.errors import backend_failure from dfe_engine.api.pagination import ( PaginatedResponse, PaginationParams, @@ -201,9 +202,13 @@ async def available_org_ids( try: org_ids = discover_available_org_ids(ch, database=database, limit=limit) except OrgIdDiscoveryError as exc: - raise HTTPException( - status_code=503, - detail={"code": "discovery_failed", "message": str(exc)}, + raise backend_failure( + 503, + "discovery_failed", + f"the tenant ids in {database} could not be read; the engine log has " + "ClickHouse's reason", + exc, + event="tenant id discovery failed", ) from exc return AvailableOrgIdsResponse(database=database, org_ids=org_ids) diff --git a/src/dfe_engine/api/v1/queries.py b/src/dfe_engine/api/v1/queries.py index 6c1825c7e..b7a0e85fe 100644 --- a/src/dfe_engine/api/v1/queries.py +++ b/src/dfe_engine/api/v1/queries.py @@ -24,6 +24,7 @@ from scalo.logger import logger from dfe_engine.api.deps import CurrentUser, Settings, require_action +from dfe_engine.api.errors import backend_failure from dfe_engine.auth.rbac_scopes import scopes_dict from dfe_engine.query.models import ( QueryOptions, @@ -198,11 +199,14 @@ def execute_view( detail={"code": "invalid_options", "message": str(exc)}, ) except ViewExecutionError as exc: - logger.error("View execution failed", label=label, error=str(exc)) - raise HTTPException( - status_code=500, - detail={"code": "query_error", "message": str(exc)}, - ) + raise backend_failure( + 500, + "query_error", + f"View '{label}' failed to execute; the engine log has ClickHouse's reason", + exc, + event="View execution failed", + label=label, + ) from exc meta = result.metadata return QueryResponse( @@ -321,9 +325,11 @@ async def query_cost_leaderboard( database=settings.clickhouse.effective_data_database, ) except Exception as exc: - logger.error("cost leaderboard query failed", error=str(exc)) - raise HTTPException( - status_code=503, - detail={"code": "not_available", "message": f"cost leaderboard unavailable: {exc}"}, - ) + raise backend_failure( + 503, + "not_available", + "cost leaderboard unavailable; the engine log has the reason", + exc, + event="cost leaderboard query failed", + ) from exc return [CostLeaderboardRow(**row) for row in rows] diff --git a/src/dfe_engine/api/v1/sampler.py b/src/dfe_engine/api/v1/sampler.py index 228ac4a2c..7476aa2ac 100644 --- a/src/dfe_engine/api/v1/sampler.py +++ b/src/dfe_engine/api/v1/sampler.py @@ -157,7 +157,14 @@ async def _submit( await check_sample_scope(sampler, req, ch, source_registry, org_ids) info = manager.submit( - _TASK_KIND, sampler.run, req, ch, source_registry, held_to=org_ids, org_ids=org_ids + _TASK_KIND, + sampler.run, + req, + ch, + source_registry, + held_to=org_ids, + reportable=(SamplerError,), + org_ids=org_ids, ) audit_resource_change( user.user_id, "sampler", req.source or req.table or req.topic or "", "executed" diff --git a/src/dfe_engine/api/v1/schemas.py b/src/dfe_engine/api/v1/schemas.py index 69511f55a..d9a3b7f31 100644 --- a/src/dfe_engine/api/v1/schemas.py +++ b/src/dfe_engine/api/v1/schemas.py @@ -29,7 +29,7 @@ require_action, ttl_settings, ) -from dfe_engine.api.errors import ErrorResponse, raise_exchange_http +from dfe_engine.api.errors import ErrorResponse, engine_message, raise_exchange_http from dfe_engine.api.pagination import ( PaginatedResponse, PaginationParams, @@ -1654,6 +1654,20 @@ def _discovery_target(source, ver, schema_registry, *, version_id: str, ch=None, return landing, match_rule, [] +def _promotion_refused(exc: Exception, code: str, source_name: str) -> HTTPException: + """The 422 for a JSON-path read: the engine's refusal as written, ClickHouse's text logged.""" + message = engine_message( + exc, + "ClickHouse could not read the source's JSON paths; the engine log has the reason", + event="JSON path read failed", + source=source_name, + ) + return HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, + detail={"code": code, "message": message}, + ) + + def _held_read_refused(exc: Exception) -> HTTPException: """The 403 for a row read that cannot be held to the caller's orgs.""" return HTTPException( @@ -1775,10 +1789,7 @@ async def discover_json_paths( except JsonPromotionScopeError as exc: raise _held_read_refused(exc) from exc except JsonPromotionError as exc: - raise HTTPException( - status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, - detail={"code": "discovery_failed", "message": str(exc)}, - ) from exc + raise _promotion_refused(exc, "discovery_failed", source_name) from exc return JsonPathsResponse( source_name=source_name, @@ -1873,10 +1884,7 @@ async def sample_source_rows( except JsonPromotionScopeError as exc: raise _held_read_refused(exc) from exc except JsonPromotionError as exc: - raise HTTPException( - status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, - detail={"code": "sample_failed", "message": str(exc)}, - ) from exc + raise _promotion_refused(exc, "sample_failed", source_name) from exc promoted = [ SampleRowsResponse.PromotedJsonField(**item) @@ -2009,10 +2017,7 @@ async def promote_field( except JsonPromotionScopeError as exc: raise _held_read_refused(exc) from exc except JsonPromotionError as exc: - raise HTTPException( - status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, - detail={"code": "discovery_failed", "message": str(exc)}, - ) from exc + raise _promotion_refused(exc, "discovery_failed", source_name) from exc path_types = {d.path: d.types for d in discovered} outcomes = build_promotion_columns( diff --git a/src/dfe_engine/api/v1/sources.py b/src/dfe_engine/api/v1/sources.py index 50fa86a3a..6757dac17 100644 --- a/src/dfe_engine/api/v1/sources.py +++ b/src/dfe_engine/api/v1/sources.py @@ -57,10 +57,12 @@ ErrorResponse, SourceCreateConflictResponse, SourceWriteConflictResponse, + backend_failure, + hide_backend_text, raise_exchange_http, ) from dfe_engine.api.pagination import PaginatedResponse, PaginationParams, apply_search, apply_sort -from dfe_engine.api.v1.apps import commit_overlay, remove_overlay +from dfe_engine.api.v1.apps import commit_overlay, metrics_unavailable, remove_overlay from dfe_engine.api.write_turn import WRITE_TURN from dfe_engine.appmgmt import ( LOADER_COMPILER, @@ -116,6 +118,20 @@ router = APIRouter(prefix="/sources", tags=["Sources"], dependencies=[WRITE_TURN]) +_CH_UNREADABLE = "ClickHouse could not be read; the engine log has the reason" + + +def _ch_unreadable(exc: Exception, source: str) -> HTTPException: + """The 503 for a deploy or plan that could not read ClickHouse's state.""" + return backend_failure( + 503, + "clickhouse_unavailable", + _CH_UNREADABLE, + exc, + event="source deploy: ClickHouse unreadable", + source=source, + ) + def _raise_save_validation_http(exc: SourceValidationError) -> NoReturn: """Map registry validation errors to HTTP responses (never 500).""" @@ -304,8 +320,13 @@ def _reconcile_apps(request: Request, user: Any, registry: Any) -> AppsSync: logger.warning(f"apps not reconciled with the sources: {detail}") return AppsSync(changes=done, error=str(detail), restart_required=_one_per_app(hints)) except Exception as exc: - logger.warning(f"apps not reconciled with the sources: {exc}") - return AppsSync(changes=done, error=str(exc), restart_required=_one_per_app(hints)) + # Anything but an HTTPException can be the deploy repo's own text. + error = hide_backend_text( + "the deploy repo could not be brought into step; the engine log has the reason", + exc, + event="apps not reconciled with the sources", + ) + return AppsSync(changes=done, error=error, restart_required=_one_per_app(hints)) # Re-rendered even when the reconcile wrote nothing: a source deploy changes # the loader's table map through the same overlay the plan found in step. hints += appconfig.render_and_report(gc, settings) @@ -335,8 +356,12 @@ async def _sync_hyperdx_source( columns=columns, ) except Exception as exc: - logger.warning(f"HyperDX not pointed at source '{source.source}': {exc}") - return None, str(exc) + return None, hide_backend_text( + "Search did not accept the source; the engine log has the reason", + exc, + event="HyperDX not pointed at the source", + source=source.source, + ) if teams is None: return None, "Search did not accept the source; see the engine log" return len(teams), None @@ -1257,10 +1282,7 @@ def plan_source_deploy( ch_client=ch_client, ) except SchemaApplyError as exc: - raise HTTPException( - status_code=503, - detail={"code": "clickhouse_unavailable", "message": str(exc)}, - ) from exc + raise _ch_unreadable(exc, name) from exc plan = plan_from_build( result, version=version_id, @@ -1515,7 +1537,9 @@ def _dry_run_ttl_change( try: live = live_tables(get_interactive_clickhouse_client(settings), db).get(source.table_name) except SchemaApplyError as exc: - return None, str(exc) + return None, hide_backend_text( + _CH_UNREADABLE, exc, event="dry run: live table unreadable", source=source.source + ) if live is None: return None, None change = ttl_change_for_table(builder, source, version_id, result, db=db, live=live.ttl) @@ -1572,10 +1596,7 @@ def _apply_source_schema( get_clickhouse_config(settings) ).get_clickhouse_client() except Exception as exc: - raise HTTPException( - status_code=503, - detail={"code": "clickhouse_unavailable", "message": str(exc)}, - ) from exc + raise _ch_unreadable(exc, name) from exc resolver = EngineResolver(client=ch, topology_setting=settings.clickhouse.topology) builder = SchemaBuilderV2( @@ -1645,10 +1666,7 @@ def _apply_source_schema( ch_client=ch, ) except SchemaApplyError as exc: - raise HTTPException( - status_code=503, - detail={"code": "clickhouse_unavailable", "message": str(exc)}, - ) from exc + raise _ch_unreadable(exc, name) from exc applied = 0 try: # Through the applier on the sensed resolver: a bare CREATE DATABASE lands @@ -1660,12 +1678,15 @@ def _apply_source_schema( execute_ddl(ch, stmt) applied += 1 except Exception as exc: - raise HTTPException( - status_code=502, - detail={ - "code": "ddl_apply_error", - "message": f"ClickHouse rejected DDL after {applied} statement(s): {exc}", - }, + raise backend_failure( + 502, + "ddl_apply_error", + f"ClickHouse rejected DDL after {applied} statement(s); the engine log has " + "ClickHouse's reason", + exc, + event="source deploy: ClickHouse rejected DDL", + source=name, + applied=applied, ) from exc # The tier grant is database-wide, so the new table is readable the moment it @@ -1762,9 +1783,7 @@ def get_source_signals( client, settings.clickhouse.effective_data_database ).source_signals(table, loaders) except MetricsUnavailableError as exc: - raise HTTPException( - status_code=503, detail={"code": "metrics_unavailable", "message": str(exc)} - ) from exc + raise metrics_unavailable(exc) from exc return SourceSignalsResponse( source=source.source, table=table, @@ -2054,7 +2073,18 @@ def _apply_bulk( registry.save_source(updated, created_by=git_author(user)) succeeded.append(name) except Exception as e: - failed.append({"source": name, "code": _failure_code(e), "error": str(e)}) + code = _failure_code(e) + # A refusal the registry composed is the caller's to read; anything else may be + # the deploy repo's own text. + error = str(e) + if code == "internal_error": + error = hide_backend_text( + "the source could not be written; the engine log has the reason", + e, + event="bulk source action failed", + source=name, + ) + failed.append({"source": name, "code": code, "error": error}) return succeeded, failed, deleted diff --git a/src/dfe_engine/api/v1/synthetic_data.py b/src/dfe_engine/api/v1/synthetic_data.py index 7e00d35b2..94c1c48ca 100644 --- a/src/dfe_engine/api/v1/synthetic_data.py +++ b/src/dfe_engine/api/v1/synthetic_data.py @@ -154,7 +154,7 @@ async def start_stream( "cancel one or raise synthetic_data.max_concurrent_streams", }, ) - info = manager.submit(_TASK_KIND, service.run_stream, body) + info = manager.submit(_TASK_KIND, service.run_stream, body, reportable=(SyntheticDataError,)) audit_resource_change( user.user_id, "synthetic-data", diff --git a/src/dfe_engine/api/v1/system.py b/src/dfe_engine/api/v1/system.py index 7c2ef880c..9447c9f72 100644 --- a/src/dfe_engine/api/v1/system.py +++ b/src/dfe_engine/api/v1/system.py @@ -30,7 +30,7 @@ get_interactive_clickhouse_client, require_action, ) -from dfe_engine.api.errors import ErrorResponse +from dfe_engine.api.errors import ErrorResponse, backend_failure, engine_message from dfe_engine.api.pagination import PaginatedResponse, PaginationParams, apply_search from dfe_engine.api.write_turn import WRITE_TURN from dfe_engine.appmgmt import DeployTarget, appconfig, routing @@ -86,6 +86,28 @@ def _gitcrud(request: Request) -> GitCrud: return gc +def _ttl_reconcile_failed(exc: Exception) -> HTTPException: + """The 502 for a stored TTL override that ClickHouse did not take.""" + return backend_failure( + 502, + "reconcile_failed", + "override stored; applying it to ClickHouse failed, and the engine log has " + "ClickHouse's reason. Sending the same value again applies it", + exc, + event="default TTL stored but not applied", + ) + + +def _cloud_error(exc: Exception) -> HTTPException: + """The 502 for a ClickHouse Cloud management-API failure.""" + message = engine_message( + exc, + "the ClickHouse Cloud management API call failed; the engine log has the reason", + event="ClickHouse Cloud management API call failed", + ) + return HTTPException(status_code=502, detail={"code": "cloud_error", "message": message}) + + # -- Response models ------------------------------------------ @@ -156,7 +178,9 @@ class SchemaStatusResponse(BaseModel): started_at: str finished_at: str duration_seconds: float - error: str = Field(description="Why the pass failed; empty when it did not.") + error: str = Field( + description="The stage that failed, its cause in the engine log; empty when none did." + ) counts: dict[str, int] = Field(default_factory=dict, description="One count per action") objects: list[SchemaObjectStatus] = Field(default_factory=list) refused: list[str] = Field( @@ -395,7 +419,7 @@ async def get_schema_status(user: CurrentUser, settings: Settings) -> SchemaStat This is the operator's record of the schema apply, replacing the completed ArgoCD Job the engine took over from. ``state`` is what readiness follows: converged or observed is ready, failed leaves the pod up and NotReady with - the cause here. + the stage that failed here and the cause in the engine log. """ from dfe_engine.schema.phase import current_state @@ -591,13 +615,7 @@ def put_retention( ) # The override is committed by now, so any ClickHouse failure has to say so rather than 500. except Exception as exc: - raise HTTPException( - status_code=502, - detail={ - "code": "reconcile_failed", - "message": f"override stored; applying it to ClickHouse failed: {exc}", - }, - ) from exc + raise _ttl_reconcile_failed(exc) from exc if outcome.sources_skipped: logger.warning( "default TTL: sources left to their next deploy", count=outcome.sources_skipped @@ -825,13 +843,7 @@ def patch_defaults( sources=sources.get_all_sources(), ) except Exception as exc: - raise HTTPException( - status_code=502, - detail={ - "code": "reconcile_failed", - "message": f"override stored; applying it to ClickHouse failed: {exc}", - }, - ) from exc + raise _ttl_reconcile_failed(exc) from exc if outcome.sources_skipped: logger.warning( "default TTL: sources left to their next deploy", count=outcome.sources_skipped @@ -1105,12 +1117,12 @@ def _deployed_tables(settings: Any, sources: list[Source]) -> dict[str, LiveTabl try: return live_tables(get_interactive_clickhouse_client(settings), database) except SchemaApplyError as exc: - raise HTTPException( - status_code=503, - detail={ - "code": "clickhouse_unavailable", - "message": f"the deployed tables could not be read: {exc}", - }, + raise backend_failure( + 503, + "clickhouse_unavailable", + "the deployed tables could not be read; the engine log has ClickHouse's reason", + exc, + event="default drift: deployed tables unreadable", ) from exc @@ -1209,9 +1221,7 @@ def _cloud_state(settings) -> CloudServiceStateResponse: try: st = CloudService(cloud).status() except CloudServiceError as exc: - raise HTTPException( - status_code=502, detail={"code": "cloud_error", "message": str(exc)} - ) from exc + raise _cloud_error(exc) from exc return CloudServiceStateResponse( configured=True, id=st.id, name=st.name, state=st.state, is_running=st.is_running ) @@ -1246,9 +1256,7 @@ def clickhouse_cloud_start(user: CurrentUser, settings: Settings): try: st = CloudService(cloud).start() except CloudServiceError as exc: - raise HTTPException( - status_code=502, detail={"code": "cloud_error", "message": str(exc)} - ) from exc + raise _cloud_error(exc) from exc return CloudServiceStateResponse( configured=True, id=st.id, name=st.name, state=st.state, is_running=st.is_running ) @@ -1273,9 +1281,7 @@ def clickhouse_cloud_stop(user: CurrentUser, settings: Settings): try: st = CloudService(cloud).stop() except CloudServiceError as exc: - raise HTTPException( - status_code=502, detail={"code": "cloud_error", "message": str(exc)} - ) from exc + raise _cloud_error(exc) from exc return CloudServiceStateResponse( configured=True, id=st.id, name=st.name, state=st.state, is_running=st.is_running ) diff --git a/src/dfe_engine/kafka/topics.py b/src/dfe_engine/kafka/topics.py index b487b063c..b77b49b9a 100644 --- a/src/dfe_engine/kafka/topics.py +++ b/src/dfe_engine/kafka/topics.py @@ -63,6 +63,16 @@ from dfe_engine.settings import DFESettings from dfe_engine.source.models import Source +# A reason recorded here reaches API callers, so the broker's own text goes to the log. +BROKER_UNREACHABLE = "broker unreachable; the engine log has the reason" +BROKER_REFUSED = "the broker refused it; the engine log has the reason" + + +def _unreachable(exc: Exception) -> str: + """The reason recorded for a broker that could not be reached, its own text logged.""" + logger.warning("Kafka broker unreachable", error=str(exc)) + return BROKER_UNREACHABLE + @dataclass class TopicSpec: @@ -595,7 +605,8 @@ def ensure_topics( result.failed = [(spec.name, f"kafka config rejected: {exc}") for spec in specs] return result except Exception as exc: - result.failed = [(spec.name, f"broker unreachable: {exc}") for spec in specs] + reason = _unreachable(exc) + result.failed = [(spec.name, reason) for spec in specs] return result for spec in specs: @@ -615,7 +626,7 @@ def ensure_topics( f"(partitions={spec.partitions}, rf={spec.replication_factor})" ) except Exception as exc: - result.failed.append((spec.name, str(exc))) + result.failed.append((spec.name, BROKER_REFUSED)) logger.error(f"Kafka topic creation failed: {spec.name} - {exc}") return result @@ -653,7 +664,8 @@ def remove_topics( result.failed = [(name, f"kafka config rejected: {exc}") for name in names] return result except Exception as exc: - result.failed = [(name, f"broker unreachable: {exc}") for name in names] + reason = _unreachable(exc) + result.failed = [(name, reason) for name in names] return result for name in names: @@ -665,7 +677,7 @@ def remove_topics( result.removed.append(name) logger.info(f"Kafka topic deleted: {name}") except Exception as exc: - result.failed.append((name, str(exc))) + result.failed.append((name, BROKER_REFUSED)) logger.error(f"Kafka topic deletion failed: {name} - {exc}") return result @@ -708,7 +720,7 @@ def topic_status( result.error = f"kafka config rejected: {exc}" return result except Exception as exc: - result.error = f"broker unreachable: {exc}" + result.error = _unreachable(exc) return result owners = sources or {} @@ -728,7 +740,8 @@ def topic_status( state.partitions, state.replication_factor = admin.describe_shape(spec.name) state.config = admin.describe_config(spec.name) except Exception as exc: - state.drift.append(f"could not be described: {exc}") + logger.warning("Kafka topic could not be described", topic=spec.name, error=str(exc)) + state.drift.append("could not be described; the engine log has the broker's reason") result.topics.append(state) continue @@ -777,7 +790,8 @@ def update_topics( result.failed = [(spec.name, f"kafka config rejected: {exc}") for spec in specs] return result except Exception as exc: - result.failed = [(spec.name, f"broker unreachable: {exc}") for spec in specs] + reason = _unreachable(exc) + result.failed = [(spec.name, reason) for spec in specs] return result for spec in specs: @@ -790,7 +804,7 @@ def update_topics( # that would find that out is skipped. drifted = _config_drift(spec, admin.describe_config(spec.name)) if spec.config else {} except Exception as exc: - result.failed.append((spec.name, str(exc))) + result.failed.append((spec.name, BROKER_REFUSED)) logger.error(f"Kafka topic could not be described: {spec.name} - {exc}") continue @@ -816,7 +830,7 @@ def update_topics( try: admin.alter_config(spec.name, drifted) except Exception as exc: - result.failed.append((spec.name, str(exc))) + result.failed.append((spec.name, BROKER_REFUSED)) logger.error(f"Kafka topic config alter failed: {spec.name} - {exc}") continue logger.info(f"Kafka topic config altered: {spec.name} ({sorted(drifted)})") @@ -829,7 +843,7 @@ def update_topics( try: admin.widen_partitions(spec.name, total=spec.partitions) except Exception as exc: - result.failed.append((spec.name, str(exc))) + result.failed.append((spec.name, BROKER_REFUSED)) logger.error(f"Kafka topic partition widening failed: {spec.name} - {exc}") continue logger.info( diff --git a/src/dfe_engine/query/executor.py b/src/dfe_engine/query/executor.py index 3669732f4..650b70977 100644 --- a/src/dfe_engine/query/executor.py +++ b/src/dfe_engine/query/executor.py @@ -22,6 +22,8 @@ import uuid from typing import Any +from scalo.logger import logger + from dfe_engine.clickhouse.quoting import BARE_REFERENCE, column_reference, quote_identifier from dfe_engine.query.catalog import RESERVED_PARAMS, ViewCatalog from dfe_engine.query.models import ( @@ -121,9 +123,11 @@ def execute( ) except Exception as e: if options.after_key is not None and _TYPE_MISMATCH.search(str(e)): + # A ViewRequestError's message reaches the caller; ClickHouse's text goes to the log. + logger.warning("view cursor does not parse", label=label, error=str(e)) raise ViewRequestError( "after_key or after_tiebreak does not parse as its column's type " - f"(a timestamp is sent as e.g. '2024-01-15T12:00:00'): {e}" + "(a timestamp is sent as e.g. '2024-01-15T12:00:00')" ) from e raise ViewExecutionError(f"Failed to execute view '{label}': {e}") from e diff --git a/src/dfe_engine/sampling/service.py b/src/dfe_engine/sampling/service.py index 2476afe06..47ea9a2c0 100644 --- a/src/dfe_engine/sampling/service.py +++ b/src/dfe_engine/sampling/service.py @@ -138,7 +138,11 @@ def check_org_scope( try: columns = ch_reader.column_names(ch, target, self._cfg.max_execution_time) except DatabaseError as exc: - raise SamplerError(f"cannot read the columns of {target}: {exc}") from exc + # ClickHouse's text goes to the log: a SamplerError's message reaches the caller. + logger.warning("sample target columns unreadable", target=target, error=str(exc)) + raise SamplerError( + f"cannot read the columns of {target}; the engine log has ClickHouse's reason" + ) from exc if ORG_ID_COLUMN not in columns: raise SampleScopeError( f"{target} has no {ORG_ID_COLUMN} column, so a sample of it cannot be " diff --git a/src/dfe_engine/schema/phase.py b/src/dfe_engine/schema/phase.py index 8cac96164..a1449ce66 100644 --- a/src/dfe_engine/schema/phase.py +++ b/src/dfe_engine/schema/phase.py @@ -15,8 +15,9 @@ topics are on the broker. Two things make it a GATE rather than the best-effort pass it replaces. A failure -leaves the engine UP and NotReady with the cause on the status route, because a -silent failure and a success look identical to every app downstream. And +leaves the engine UP and NotReady, with the stage that failed on the status route +and the cause in the engine log, because a silent failure and a success look +identical to every app downstream. And ClickHouse or the broker merely not being up yet is not a failure: the phase retries for a bounded window first, which is what a stack whose datastore starts in the same wave needs. @@ -69,6 +70,24 @@ class DeadLetterPathError(Exception): STATE_OBSERVED: 4, } +# The state is served to every signed-in caller, so a backend's own error text stays +# in the log and the state names only the stage that failed. +_FAILED_STAGES: tuple[tuple[type[Exception], str], ...] = ( + (ManifestApplyError, "ClickHouse refused or did not answer the manifest apply"), + (SchemaPlanError, "the schema manifest could not be planned"), + (SchemaLockError, "the schema lock could not be read or written"), + (LedgerError, "the migration ledger could not be read or written"), +) +_SEE_LOG = "the engine log has the cause" + + +def _failed_stage(exc: Exception) -> str: + """The status route's account of a failed pass: the stage, never the backend's text.""" + for kind, stage in _FAILED_STAGES: + if isinstance(exc, kind): + return f"{stage}; {_SEE_LOG}" + return f"unexpected failure; {_SEE_LOG}" + @dataclass class SchemaBootstrapState: @@ -411,8 +430,9 @@ def run_bootstrap( """Bring this deployment's ClickHouse and topic set to the pinned manifest. Returns the state, which is also what the status route and the readiness - check read. Never raises: a failure is reported as state ``failed`` with the - cause, so the pod stays up for an operator to read it off the API. + check read. Never raises: a failure is reported as state ``failed`` naming the + stage that failed, with the cause in the engine log, so the pod stays up for an + operator to find it. """ ch = settings.clickhouse started = datetime.now(UTC) @@ -456,13 +476,13 @@ def run_bootstrap( ) except (ManifestApplyError, SchemaPlanError, SchemaLockError, LedgerError) as exc: state.state = STATE_FAILED - state.error = str(exc) + state.error = _failed_stage(exc) logger.error("schema bootstrap failed; the engine stays up and NotReady", error=str(exc)) _finish(state, started) return state except Exception as exc: state.state = STATE_FAILED - state.error = f"unexpected failure: {exc}" + state.error = _failed_stage(exc) logger.exception("schema bootstrap failed; the engine stays up and NotReady") _finish(state, started) return state @@ -490,7 +510,7 @@ def run_bootstrap( plan, settings ) except Exception as exc: # a broker fault must not take the ClickHouse apply with it - state.topics_skipped = f"topic bootstrap failed: {exc}" + state.topics_skipped = f"topic bootstrap failed; {_SEE_LOG}" logger.warning("bootstrap topics not created", error=str(exc)) try: diff --git a/src/dfe_engine/schema/retention.py b/src/dfe_engine/schema/retention.py index 3605cce34..327c7b04f 100644 --- a/src/dfe_engine/schema/retention.py +++ b/src/dfe_engine/schema/retention.py @@ -56,6 +56,9 @@ # A table's engine is fixed when it is created; changing it means copying the data out. ENGINE_NEEDS_REBUILD = "needs a table rebuild; applies to new tables only" +# An outcome's reason reaches API callers, so ClickHouse's own text stays in the log. +CLICKHOUSE_REFUSED = "ClickHouse could not be read or refused the change; the engine log has why" + # What one source's table can refuse: the build, its types, its DDL or the server. _TABLE_ERRORS = ( SchemaApplyError, @@ -341,7 +344,7 @@ def _apply_deployed( source=source.source, status="failed", table=f"{database}.{source.table_name}", - reason=str(exc), + reason=CLICKHOUSE_REFUSED, ) for source in sources ] @@ -362,9 +365,9 @@ def _apply_one(tables: SourceTables, source: Source, live: LiveTable | None) -> kept = tables.header_nullability_kept(source, source.current, applied.columns) except _TABLE_ERRORS as exc: logger.warning(f"{target}: table defaults not applied: {exc}") - return SourceLiveOutcome( - source=source.source, status="failed", table=target, reason=str(exc) - ) + # The build, type and DDL errors are about the source's own schema; the server's is not. + reason = CLICKHOUSE_REFUSED if isinstance(exc, SchemaApplyError) else str(exc) + return SourceLiveOutcome(source=source.source, status="failed", table=target, reason=reason) not_applied: list[FieldNotApplied] = [] if applied.change.ttl_skipped: not_applied.append(FieldNotApplied("ttl_days", applied.change.ttl_skipped)) diff --git a/tests/unit/test_api/test_gitops_unavailable.py b/tests/unit/test_api/test_gitops_unavailable.py index 6fc8af096..54dc5a0f9 100644 --- a/tests/unit/test_api/test_gitops_unavailable.py +++ b/tests/unit/test_api/test_gitops_unavailable.py @@ -19,6 +19,7 @@ from prometheus_client.parser import text_string_to_metric_families from scalo.metrics import create_metrics +from dfe_engine.api.errors import GITOPS_UNAVAILABLE_MESSAGE from dfe_engine.gitcrud import GitCrud, default_registry from dfe_engine.gitops.metrics import GitopsMetrics from dfe_engine.gitops.repo import GitopsRepo @@ -82,7 +83,7 @@ def test_an_unreachable_deploy_repo_answers_503_with_retry_after( body = resp.json() assert body["code"] == "service_unavailable" assert body["context"] == {"service": "gitops", "remote": remote} - assert "gitops deploy repo" in body["message"] + assert body["message"] == GITOPS_UNAVAILABLE_MESSAGE labels = {"op": "fetch", "outcome": "exhausted"} assert _sample(manager, "gitops_write_retries_total", labels) == 1 diff --git a/tests/unit/test_api/test_no_backend_error_text.py b/tests/unit/test_api/test_no_backend_error_text.py new file mode 100644 index 000000000..7b1754980 --- /dev/null +++ b/tests/unit/test_api/test_no_backend_error_text.py @@ -0,0 +1,477 @@ +# Project: dfe-engine +# File: tests/unit/test_api/test_no_backend_error_text.py +# Purpose: A backend's own error text reaches the engine log, never a response +# Language: Python +# +# License: BUSL-1.1 +# Copyright: (c) 2026 HYPERI PTY LIMITED + +"""Every route that reports a ClickHouse, Kafka, git or HTTP failure keeps the backend's text out. + +ClickHouse's error text can carry statement fragments, user names and password +hashes, and these routes reach viewers through the console. Each test makes the +backend fail with :data:`SENTINEL`, then checks the response carries none of it +and the engine log carries all of it. +""" + +from types import SimpleNamespace + +import pytest +from clickhouse_connect.driver.exceptions import DatabaseError +from fastapi import FastAPI +from fastapi.testclient import TestClient +from scalo.resilience import ServiceUnavailable + +from dfe_engine.api.deps import get_clickhouse_client +from dfe_engine.api.errors import ( + GITOPS_UNAVAILABLE_MESSAGE, + SERVICE_UNAVAILABLE_MESSAGE, + engine_message, + install_exception_handlers, +) +from dfe_engine.api.v1 import orgs as orgs_module +from dfe_engine.api.v1 import sources as sources_module +from dfe_engine.api.v1 import system as system_module +from dfe_engine.api.v1.apps import metrics_unavailable +from dfe_engine.api.v1.schemas import _promotion_refused +from dfe_engine.appmgmt import MetricsUnavailableError +from dfe_engine.auth.models import AuthContext +from dfe_engine.clickhouse.clickhouse_manager import ClickHouseManager +from dfe_engine.clickhouse.cloud import CloudServiceError +from dfe_engine.gitcrud import GitCrud, default_registry +from dfe_engine.gitops.repo import GitopsRepo, GitopsUnavailableError +from dfe_engine.kafka.topics import ( + BROKER_REFUSED, + BROKER_UNREACHABLE, + TopicSpec, + ensure_topics, + topic_status, +) +from dfe_engine.orgs.available_ids import OrgIdDiscoveryError +from dfe_engine.query.executor import ViewExecutionError +from dfe_engine.sampling import SamplerError +from dfe_engine.sampling.models import SampleBackend, SampleRequest +from dfe_engine.sampling.service import Sampler +from dfe_engine.schema import phase +from dfe_engine.schema.applier import SchemaApplyError +from dfe_engine.schema.manifest_applier import ManifestApplyError +from dfe_engine.schema.retention import CLICKHOUSE_REFUSED, apply_pinned_defaults +from dfe_engine.services.schema.json_promotion_service import JsonPromotionError +from dfe_engine.source.registry import SourceNotFoundError + +FAKE_HASH = "9f86d081884c7d659a2feaa0c55ad015a3bf4f1b2b0b822cd15d6c15b0f00a08" +SENTINEL = f"Code: 516. DB::Exception: ALTER USER dfe_org_acme IDENTIFIED WITH sha256_hash BY '{FAKE_HASH}'" + + +def assert_hidden(text: str) -> None: + """The response text carries no part of the backend's error.""" + assert "IDENTIFIED" not in text + assert FAKE_HASH not in text + + +def assert_logged(events: list[dict]) -> None: + """Some log line carries the backend's error whole, in its message or its error field.""" + assert any(FAKE_HASH in f"{e.get('event', '')} {e.get('error', '')}" for e in events), events + + +def _refuse(*_args, **_kwargs): + raise RuntimeError(SENTINEL) + + +def _chained(kind: type[Exception], message: str) -> Exception: + """``kind(message)`` raised from a backend error carrying the sentinel, as the engine raises it.""" + try: + raise RuntimeError(SENTINEL) + except RuntimeError as backend: + try: + raise kind(f"{message}: {backend}") from backend + except kind as wrapped: + return wrapped + + +class TestCentralHandlers: + @staticmethod + def _client() -> TestClient: + app = FastAPI() + install_exception_handlers(app) + + @app.get("/dead") + def _dead(): + raise ServiceUnavailable(f"ClickHouse unreachable after 60s (3 attempts): {SENTINEL}") + + @app.get("/gitops") + def _gitops(): + raise GitopsUnavailableError(f"deploy repo unreachable after 30s: {SENTINEL}") + + return TestClient(app, raise_server_exceptions=False) + + def test_an_exhausted_backend_answers_503_without_its_text(self, audit_events): + resp = self._client().get("/dead") + + assert resp.status_code == 503 + assert resp.json()["message"] == SERVICE_UNAVAILABLE_MESSAGE + assert_hidden(resp.text) + assert_logged(audit_events) + + def test_an_unreachable_deploy_repo_answers_503_without_its_text(self, audit_events): + resp = self._client().get("/gitops") + + assert resp.status_code == 503 + assert resp.json()["message"] == GITOPS_UNAVAILABLE_MESSAGE + assert resp.headers["Retry-After"] + assert_hidden(resp.text) + assert_logged(audit_events) + + +class TestEngineMessage: + def test_a_message_the_engine_composed_is_kept(self): + refusal = CloudServiceError("CH Cloud service 'dev' not found. Available: prod (running)") + + assert engine_message(refusal, "fallback", event="e") == str(refusal) + + def test_a_message_raised_from_a_backend_is_replaced_and_logged(self, audit_events): + failure = _chained(CloudServiceError, "GET /organizations failed") + + assert engine_message(failure, "fallback", event="e") == "fallback" + assert_logged(audit_events) + + +class TestQueries: + def test_a_failed_view_names_the_view_not_clickhouses_text( + self, app, client, viewer_headers, audit_events + ): + class _FailingExecutor: + def execute(self, **_kwargs): + raise ViewExecutionError(f"Failed to execute view 'x': {SENTINEL}") + + app.state.view_executor = _FailingExecutor() + + resp = client.post( + "/api/v1/queries/views/analytics/events/execute", headers=viewer_headers, json={} + ) + + assert resp.status_code == 500, resp.text + assert resp.json()["code"] == "query_error" + assert_hidden(resp.text) + assert_logged(audit_events) + + def test_the_cost_leaderboard_hides_why_it_is_unavailable( + self, client, viewer_headers, monkeypatch, audit_events + ): + monkeypatch.setattr(ClickHouseManager, "get_instance", _refuse) + + resp = client.get("/api/v1/queries/cost-leaderboard", headers=viewer_headers) + + assert resp.status_code == 503, resp.text + assert_hidden(resp.text) + assert_logged(audit_events) + + +class TestDiscovery: + def test_a_connection_failure_is_503_without_clickhouses_text( + self, app, client, admin_headers, audit_events + ): + app.state.connection_registry = SimpleNamespace(get_client=_refuse) + + resp = client.get("/api/v1/discovery/databases", headers=admin_headers) + + assert resp.status_code == 503, resp.text + assert resp.json()["code"] == "connection_error" + assert_hidden(resp.text) + assert_logged(audit_events) + + @pytest.mark.parametrize( + "path", + [ + "/api/v1/discovery/databases", + "/api/v1/discovery/tables", + "/api/v1/discovery/tables/events/columns", + ], + ) + def test_a_query_failure_is_500_without_clickhouses_text( + self, app, client, admin_headers, audit_events, path + ): + failing = SimpleNamespace(query=_refuse) + app.state.connection_registry = SimpleNamespace(get_client=lambda _name: failing) + + resp = client.get(path, headers=admin_headers) + + assert resp.status_code == 500, resp.text + assert resp.json()["code"] == "query_error" + assert_hidden(resp.text) + assert_logged(audit_events) + + +class TestOrgs: + def test_tenant_discovery_hides_clickhouses_text( + self, app, client, admin_headers, monkeypatch, audit_events + ): + def _discovery_fails(*_args, **_kwargs): + raise _chained(OrgIdDiscoveryError, "cannot read the tenant ids in dfe") + + app.dependency_overrides[get_clickhouse_client] = lambda: object() + monkeypatch.setattr(orgs_module, "discover_available_org_ids", _discovery_fails) + + resp = client.get("/api/v1/orgs/available-ids", headers=admin_headers) + + assert resp.status_code == 503, resp.text + assert resp.json()["code"] == "discovery_failed" + assert_hidden(resp.text) + assert_logged(audit_events) + + +class TestSystem: + def test_a_stored_ttl_clickhouse_refused_says_so_without_its_text( + self, app, client, admin_headers, monkeypatch, tmp_path, audit_events + ): + app.state.gitcrud = GitCrud( + GitopsRepo(local_path=str(tmp_path / "deploy"), push=False), default_registry() + ) + monkeypatch.setattr(system_module, "get_clickhouse_client", lambda _settings: object()) + monkeypatch.setattr(system_module, "reconcile_default_ttl", _refuse) + + put = client.put( + "/api/v1/system/retention", headers=admin_headers, json={"default_ttl_days": 30} + ) + patch = client.patch( + "/api/v1/system/defaults", headers=admin_headers, json={"ttl_days": 31} + ) + + for resp in (put, patch): + assert resp.status_code == 502, resp.text + assert resp.json()["code"] == "reconcile_failed" + assert "override stored" in resp.json()["message"] + assert_hidden(resp.text) + assert_logged(audit_events) + + def test_unreadable_deployed_tables_hide_clickhouses_text(self, monkeypatch, audit_events): + def _unreadable(*_args, **_kwargs): + raise SchemaApplyError(f"could not read ClickHouse state: {SENTINEL}") + + monkeypatch.setattr(system_module, "get_interactive_clickhouse_client", lambda _s: None) + monkeypatch.setattr(system_module, "live_tables", _unreadable) + settings = SimpleNamespace(clickhouse=SimpleNamespace(effective_data_database="dfe")) + + with pytest.raises(system_module.HTTPException) as raised: + system_module._deployed_tables(settings, [SimpleNamespace(deployed_version="1")]) + + assert raised.value.status_code == 503 + assert_hidden(str(raised.value.detail)) + assert_logged(audit_events) + + def test_a_cloud_api_failure_hides_its_text_and_keeps_the_engines(self, audit_events): + transport = system_module._cloud_error(_chained(CloudServiceError, "GET /services failed")) + refusal = system_module._cloud_error(CloudServiceError("CH Cloud service 'dev' not found")) + + assert transport.status_code == 502 + assert_hidden(str(transport.detail)) + assert refusal.detail["message"] == "CH Cloud service 'dev' not found" + assert_logged(audit_events) + + def test_a_failed_schema_pass_names_the_stage_not_clickhouses_text( + self, api_settings, client, viewer_headers, monkeypatch, audit_events + ): + def _no_answer(*_args, **_kwargs): + raise _chained(ManifestApplyError, "ClickHouse did not answer") + + monkeypatch.setattr(phase, "_connect", _no_answer) + bootstrap = api_settings.model_copy( + update={ + "clickhouse": api_settings.clickhouse.model_copy(update={"bootstrap_tables": True}) + } + ) + try: + state = phase.run_bootstrap(settings=bootstrap, wait_seconds=0) + resp = client.get("/api/v1/system/schema", headers=viewer_headers) + finally: + phase._set_state(phase.SchemaBootstrapState()) + + assert state.state == phase.STATE_FAILED + assert resp.status_code == 200, resp.text + assert resp.json()["state"] == "failed" + assert "manifest apply" in resp.json()["error"] + assert_hidden(resp.text) + assert_logged(audit_events) + + def test_applying_defaults_reports_clickhouse_without_its_text(self, audit_events): + source = SimpleNamespace(source="s", table_name="s", deployed_version="2", current="2") + settings = SimpleNamespace(clickhouse=SimpleNamespace(effective_data_database="dfe")) + + (outcome,) = apply_pinned_defaults(_refuse, settings=settings, sources=[source]) + + assert outcome.status == "failed" + assert outcome.reason == CLICKHOUSE_REFUSED + assert_logged(audit_events) + + +class _Admin: + """A broker admin that refuses whichever call ``refuse`` names.""" + + def __init__(self, refuse: str) -> None: + self._refuse = refuse + + def list_topic_names(self) -> set[str]: + if self._refuse == "list": + raise RuntimeError(SENTINEL) + return set() + + def create(self, *_args, **_kwargs) -> None: + raise RuntimeError(SENTINEL) + + +class TestKafkaTopics: + _SPECS = [TopicSpec(name="acme_land", partitions=3, replication_factor=1)] + + def test_a_refused_create_records_no_broker_text(self, audit_events): + outcome = ensure_topics(self._SPECS, admin=_Admin("create")) + + assert outcome.failed == [("acme_land", BROKER_REFUSED)] + assert_logged(audit_events) + + def test_an_unreachable_broker_records_no_broker_text(self, audit_events): + ensured = ensure_topics(self._SPECS, admin=_Admin("list")) + status = topic_status(self._SPECS, admin=_Admin("list")) + + assert ensured.failed == [("acme_land", BROKER_UNREACHABLE)] + assert status.error == BROKER_UNREACHABLE + assert_logged(audit_events) + + +class TestSources: + def test_a_failed_app_reconcile_answers_502_without_the_repos_text( + self, app, client, admin_headers, monkeypatch, audit_events + ): + app.state.gitcrud = object() + monkeypatch.setattr(sources_module.derived, "plan", _refuse) + + resp = client.post("/api/v1/sources/reconcile-apps", headers=admin_headers) + + assert resp.status_code == 502, resp.text + assert resp.json()["code"] == "reconcile_failed" + assert_hidden(resp.text) + assert_logged(audit_events) + + async def test_a_refused_hyperdx_source_hides_its_text(self, monkeypatch, audit_events): + from dfe_engine.hyperdx import sources as hyperdx_sources + + async def _refused(*_args, **_kwargs): + raise RuntimeError(SENTINEL) + + monkeypatch.setattr(hyperdx_sources, "ensure_source", _refused) + request = SimpleNamespace( + app=SimpleNamespace(state=SimpleNamespace(hyperdx_client=object())) + ) + source = SimpleNamespace(source="acme", table_name="acme") + + teams, error = await sources_module._sync_hyperdx_source(request, source, "dfe", ["a"]) + + assert teams is None + assert_hidden(error) + assert_logged(audit_events) + + def test_a_dry_run_hides_why_the_live_table_was_unreadable(self, monkeypatch, audit_events): + def _unreadable(*_args, **_kwargs): + raise SchemaApplyError(f"could not read ClickHouse state: {SENTINEL}") + + monkeypatch.setattr(sources_module, "get_interactive_clickhouse_client", lambda _s: None) + monkeypatch.setattr(sources_module, "live_tables", _unreadable) + source = SimpleNamespace(source="acme", table_name="acme", deployed_version="1") + settings = SimpleNamespace(clickhouse=SimpleNamespace(effective_data_database="dfe")) + + change, error = sources_module._dry_run_ttl_change(None, source, "1", None, settings) + + assert change is None + assert_hidden(error) + assert_logged(audit_events) + + def test_an_unreadable_clickhouse_on_deploy_is_503_without_its_text(self, audit_events): + raised = sources_module._ch_unreadable( + SchemaApplyError(f"could not read ClickHouse state: {SENTINEL}"), "acme" + ) + + assert raised.status_code == 503 + assert raised.detail["code"] == "clickhouse_unavailable" + assert_hidden(str(raised.detail)) + assert_logged(audit_events) + + def test_a_bulk_failure_keeps_a_refusal_and_hides_the_repos_text(self, audit_events): + class _Registry: + def source_exists(self, name: str) -> bool: + return False + + def delete_source(self, name: str, **_kwargs) -> None: + if name == "gone": + raise SourceNotFoundError("Source 'gone' not found") + raise RuntimeError(SENTINEL) + + body = SimpleNamespace(action="delete", sources=["acme", "gone"]) + user = AuthContext(user_id="admin") + + _, failed, _ = sources_module._apply_bulk(body, {}, user, _Registry()) + + by_source = {entry["source"]: entry for entry in failed} + assert by_source["gone"]["error"] == "Source 'gone' not found" + assert by_source["acme"]["code"] == "internal_error" + assert_hidden(by_source["acme"]["error"]) + assert_logged(audit_events) + + def test_unreadable_telemetry_is_503_without_clickhouses_text(self, audit_events): + raised = metrics_unavailable(MetricsUnavailableError(f"otel query 'up' failed: {SENTINEL}")) + + assert raised.status_code == 503 + assert raised.detail["code"] == "metrics_unavailable" + assert_hidden(str(raised.detail)) + assert_logged(audit_events) + + +class TestSchemas: + def test_a_json_path_read_clickhouse_refused_hides_its_text(self, audit_events): + raised = _promotion_refused( + _chained(JsonPromotionError, "JSON path discovery failed"), "discovery_failed", "acme" + ) + + assert raised.status_code == 422 + assert_hidden(str(raised.detail)) + assert_logged(audit_events) + + def test_a_refused_json_path_keeps_the_engines_message(self): + raised = _promotion_refused( + JsonPromotionError("Illegal JSON path: 'a b'"), "discovery_failed", "acme" + ) + + assert raised.detail["message"] == "Illegal JSON path: 'a b'" + + +class TestSampler: + def test_an_undescribable_target_hides_clickhouses_text(self, audit_events): + sampler = Sampler( + SimpleNamespace(max_execution_time=5), + None, + SimpleNamespace(effective_data_database="dfe"), + ) + ch = SimpleNamespace(query=lambda *_a, **_k: (_ for _ in ()).throw(DatabaseError(SENTINEL))) + req = SampleRequest(backend=SampleBackend.CLICKHOUSE, table="dfe.events") + + with pytest.raises(SamplerError) as raised: + sampler.check_org_scope(req, ch, None, ["acme"]) + + assert_hidden(str(raised.value)) + assert_logged(audit_events) + + +class TestOidcCallback: + def test_a_refused_code_exchange_hides_the_idps_text(self, app, client, audit_events): + class _RefusingRp: + def has_provider(self, provider: str) -> bool: + return True + + async def handle_callback(self, provider, request): + raise RuntimeError(SENTINEL) + + app.state.oidc_rp = _RefusingRp() + + resp = client.get("/api/v1/auth/oidc/acme/callback") + + assert resp.status_code == 401, resp.text + assert_hidden(resp.text) + assert_logged(audit_events) diff --git a/tests/unit/test_api/test_task_manager.py b/tests/unit/test_api/test_task_manager.py index 877b20b6d..117d211c6 100644 --- a/tests/unit/test_api/test_task_manager.py +++ b/tests/unit/test_api/test_task_manager.py @@ -13,7 +13,7 @@ import pytest -from dfe_engine.api.task_manager import TaskManager, TaskStatus +from dfe_engine.api.task_manager import TASK_FAILED_MESSAGE, TaskManager, TaskStatus @pytest.fixture @@ -63,17 +63,35 @@ async def weird_result(*, task): json.dumps(result.model_dump(mode="json")) @pytest.mark.asyncio - async def test_task_failure_records_error(self, manager: TaskManager): + async def test_a_reportable_failure_keeps_its_message(self, manager: TaskManager): async def fail(*, task): raise ValueError("boom") - info = manager.submit("test:fail", fail) + info = manager.submit("test:fail", fail, reportable=(ValueError,)) await asyncio.sleep(0.1) result = manager.get(info.id) assert result is not None assert result.status == TaskStatus.FAILED - assert "boom" in result.error + assert result.error == "boom" + assert result.message == "Failed: boom" + + @pytest.mark.asyncio + async def test_any_other_failure_reads_generic_and_goes_to_the_log( + self, manager: TaskManager, audit_events: list[dict] + ): + async def fail(*, task): + raise RuntimeError("ALTER USER x IDENTIFIED WITH sha256_hash BY 'deadbeef'") + + info = manager.submit("test:fail", fail, reportable=(ValueError,)) + await asyncio.sleep(0.1) + + result = manager.get(info.id) + assert result is not None + assert result.status == TaskStatus.FAILED + assert result.error == TASK_FAILED_MESSAGE + assert "IDENTIFIED" not in result.message + assert any("deadbeef" in str(event.get("error", "")) for event in audit_events) @pytest.mark.asyncio async def test_task_cancellation(self, manager: TaskManager): diff --git a/tests/unit/test_hyperdx/test_sources.py b/tests/unit/test_hyperdx/test_sources.py index c37764b8d..af014e0ef 100644 --- a/tests/unit/test_hyperdx/test_sources.py +++ b/tests/unit/test_hyperdx/test_sources.py @@ -320,7 +320,8 @@ async def test_deploy_survives_a_raising_client(self): ) assert teams is None - assert "connection refused" in error + assert "Search did not accept the source" in error + assert "connection refused" not in error async def test_deployment_without_hyperdx_reports_nothing(self): teams, error = await _sync_hyperdx_source( diff --git a/tests/unit/test_kafka/test_topics.py b/tests/unit/test_kafka/test_topics.py index 4535aa392..cfe369aff 100644 --- a/tests/unit/test_kafka/test_topics.py +++ b/tests/unit/test_kafka/test_topics.py @@ -6,6 +6,7 @@ from dfe_engine.kafka.contract import KafkaContractError from dfe_engine.kafka.topics import ( + BROKER_UNREACHABLE, TopicAdmin, TopicSpec, admin_config, @@ -217,7 +218,7 @@ def test_an_unreachable_broker_fails_every_spec_without_raising(self): admin=_UnreachableAdmin(), ) assert [name for name, _ in result.failed] == ["filebeat_land", "filebeat_load"] - assert "connection refused" in result.failed[0][1] + assert result.failed[0][1] == BROKER_UNREACHABLE def test_dry_run_creates_nothing(self): admin = _FakeAdmin() @@ -406,7 +407,7 @@ def test_an_unreachable_broker_fails_every_name_without_raising(self): result = remove_topics(["filebeat_land"], admin=_UnreachableAdmin()) assert [name for name, _ in result.failed] == ["filebeat_land"] - assert "connection refused" in result.failed[0][1] + assert result.failed[0][1] == BROKER_UNREACHABLE def test_dry_run_deletes_nothing(self): admin = _FakeAdmin(present=["t"]) @@ -471,7 +472,7 @@ def test_an_unreachable_broker_is_not_reported_as_missing_topics(self): result = topic_status([TopicSpec("filebeat_land", 3, 1)], admin=_UnreachableAdmin()) assert not result.reachable - assert "connection refused" in result.error + assert result.error == BROKER_UNREACHABLE assert result.topics == [] def test_it_carries_the_owning_source(self): diff --git a/tests/unit/test_query/test_executor.py b/tests/unit/test_query/test_executor.py index 5d48c9ae3..047613688 100644 --- a/tests/unit/test_query/test_executor.py +++ b/tests/unit/test_query/test_executor.py @@ -419,6 +419,16 @@ def test_a_cursor_that_does_not_parse_as_its_column_is_a_request_error(self): with pytest.raises(ViewRequestError, match="does not parse as its column's type"): _run_page(QueryOptions(order_by="n", after_key="x"), error=mismatch) + def test_the_request_error_carries_none_of_clickhouses_text(self): + """A ViewRequestError's message reaches the caller; ClickHouse's text does not.""" + mismatch = Exception( + "Code: 53. TYPE_MISMATCH near ALTER USER u IDENTIFIED WITH sha256_hash BY 'c0ffee'" + ) + with pytest.raises(ViewRequestError) as raised: + _run_page(QueryOptions(order_by="n", after_key="x"), error=mismatch) + assert "IDENTIFIED" not in str(raised.value) + assert "c0ffee" not in str(raised.value) + def test_a_type_mismatch_without_a_cursor_stays_an_execution_error(self): mismatch = Exception("Code: 53. DB::Exception: TYPE_MISMATCH") with pytest.raises(ViewExecutionError) as raised: diff --git a/tests/unit/test_schema/test_dead_letter_topics.py b/tests/unit/test_schema/test_dead_letter_topics.py index 5d368b6d2..c51af6589 100644 --- a/tests/unit/test_schema/test_dead_letter_topics.py +++ b/tests/unit/test_schema/test_dead_letter_topics.py @@ -128,7 +128,7 @@ def test_a_missing_topic_is_created_and_then_on_the_broker(self, plan, declared) def test_a_topic_the_broker_refuses_fails_the_pass_by_name(self, plan): broker = _Broker(refuse={"dfe_loader_dlq"}) - with pytest.raises(DeadLetterPathError, match="dfe_loader_dlq: TOPIC_AUTHORIZATION_FAILED"): + with pytest.raises(DeadLetterPathError, match="dfe_loader_dlq: the broker refused it"): require_dead_letter_topics(plan, _settings(), wait_seconds=0, admin=broker) def test_an_unreachable_broker_fails_the_pass(self, plan):