From e7ddbfcbc9d8713bbaf60b66fd447a254ba0f811 Mon Sep 17 00:00:00 2001 From: Derek Date: Sat, 10 Oct 2026 18:20:05 +1100 Subject: [PATCH] fix: keep backend error text out of API responses ClickHouse's error text can carry statement fragments, user names and password hashes, and routes across the API handed it straight back to the caller, viewers included. Kafka, the deploy repo and the IdP leaked the same way. - A route reporting a backend failure answers a fixed message naming what failed, and the backend's text goes to the engine log. One helper in api/errors.py does it, and the CH RBAC reconcile now uses it too. - The 503 for an exhausted ClickHouse or deploy repo no longer repeats scalo's message, which ends with the backend's last error verbatim. - Background tasks log every failure. Readers see a task's own message only for the sampler's and synthetic data's refusals; anything else reads "The task failed". - Covered: view execution, cost leaderboard, discovery, tenant id discovery, TTL and defaults reconcile, default drift, ClickHouse Cloud, source plan, deploy and dry run, source signals, app metrics, app reconcile, bulk source actions, HyperDX source sync, JSON path reads, the sampler scope check, the Kafka topic routes, schema status and the OIDC callback. - Schema status names the stage that failed; the cause is in the engine log, which the CLI prints too. - Refusals the engine writes about the caller's own input keep their text: cursor and option validation, catalogue and schema build errors, Kafka contract errors, JSON path validation. --- openapi-spec/openapi.json | 6 +- src/dfe_engine/api/errors.py | 87 +++- src/dfe_engine/api/task_manager.py | 23 +- src/dfe_engine/api/v1/apps.py | 25 +- src/dfe_engine/api/v1/discovery.py | 27 +- src/dfe_engine/api/v1/governance.py | 22 +- src/dfe_engine/api/v1/kafka_topics.py | 2 +- src/dfe_engine/api/v1/oidc_login.py | 8 +- src/dfe_engine/api/v1/orgs.py | 11 +- src/dfe_engine/api/v1/queries.py | 26 +- src/dfe_engine/api/v1/sampler.py | 9 +- src/dfe_engine/api/v1/schemas.py | 31 +- src/dfe_engine/api/v1/sources.py | 86 +++- src/dfe_engine/api/v1/synthetic_data.py | 2 +- src/dfe_engine/api/v1/system.py | 70 +-- src/dfe_engine/kafka/topics.py | 34 +- src/dfe_engine/query/executor.py | 6 +- src/dfe_engine/sampling/service.py | 6 +- src/dfe_engine/schema/phase.py | 34 +- src/dfe_engine/schema/retention.py | 11 +- .../unit/test_api/test_gitops_unavailable.py | 3 +- .../test_api/test_no_backend_error_text.py | 477 ++++++++++++++++++ tests/unit/test_api/test_task_manager.py | 26 +- tests/unit/test_hyperdx/test_sources.py | 3 +- tests/unit/test_kafka/test_topics.py | 7 +- tests/unit/test_query/test_executor.py | 10 + .../test_schema/test_dead_letter_topics.py | 2 +- 27 files changed, 891 insertions(+), 163 deletions(-) create mode 100644 tests/unit/test_api/test_no_backend_error_text.py 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):