diff --git a/docs/model-import.md b/docs/model-import.md index 51ca951..50a76d8 100644 --- a/docs/model-import.md +++ b/docs/model-import.md @@ -353,7 +353,38 @@ change. There is no rollback, no compensation, no resume and no reconciliation. A failed import stays failed; trying again is a new upload and a new row. -## 15. How this contract is verified +## 15. What a failure looks like in the logs + +Every failure — a row that cannot be claimed, a gate that rejected the bundle, an MLflow that refused +the connection, a verdict that could not be written — goes through the one channel in +`laborious/utils/models/import_errors.py`, which produces exactly two `logfmt` lines: the technical +one, whose message +carries the traceback (or the failure chain, root cause first, for anything that crossed an activity +boundary), and the user-facing sentence the notification logs. **Both carry the same fields**, so one +query returns the pair: + +| Field | What it holds | +|---|---| +| `import_run_id` | the record's primary key — the id a user quotes to support | +| `import_workflow_id` | the Temporal workflow id | +| `import_step` | the step that failed, or `not_claimable` / `status_write` | +| `import_code` | the code of § 8, the same one written to the row | +| `error_type` | the failing exception's own type — `MlflowException`, not the `ApplicationError` wrapping it | +| `error_summary` | one bounded line of what it said | +| `import_gate` | the gate that rejected the bundle, when one did | +| `import_verdict_not_recorded` | the verdict a failed terminal write could not store | + +The workflow metadata every other line already carries (`pod_id`, `runtime`, `model_id`, `model_name`, +`workflow_name`, `schedule_name`) is kept alongside them. + +A failure at any pipeline step is reported from the workflow's `except`, before cleanup and before the +terminal write, so the log reads in the order things happened and the line's step can never disagree +with the row's. That report never changes the outcome: its own failure is swallowed, and the failure it +was reporting is re-raised untouched. The Temporal SDK still prints its own traceback for each failed +attempt — that is the per-attempt record, and it is not structured; the channel's line is the one to +query. + +## 16. How this contract is verified Two suites, and they prove different things. diff --git a/e2e/import_harness.py b/e2e/import_harness.py index 517f802..8d4b99d 100644 --- a/e2e/import_harness.py +++ b/e2e/import_harness.py @@ -118,6 +118,7 @@ IMPORT_ACTIVITY_NAMES = ( 'record_import_names', 'record_import_terminal_status', 'report_import_status_write_failure', + 'report_import_pipeline_failure', 'download_import_bundle', 'open_import_bundle', 'create_import_experiment', diff --git a/e2e/test_import_model_digest.py b/e2e/test_import_model_digest.py index adcc88b..d3a6ee0 100644 --- a/e2e/test_import_model_digest.py +++ b/e2e/test_import_model_digest.py @@ -25,8 +25,8 @@ import pytest from temporalio.client import WorkflowFailureError from e2e.import_harness import IMPORT_BUCKET, failure_text -from laborious.activities.model_import_errors import IMPORT_FAILURE_CATALOG from laborious.utils.bundle.steps import ImportStep +from laborious.utils.models.import_errors import IMPORT_FAILURE_CATALOG pytestmark = [pytest.mark.import_e2e, pytest.mark.asyncio] diff --git a/e2e/test_import_model_gates.py b/e2e/test_import_model_gates.py index be9ab0e..fd3e29d 100644 --- a/e2e/test_import_model_gates.py +++ b/e2e/test_import_model_gates.py @@ -30,8 +30,8 @@ from e2e.import_harness import ( IMPORT_PREFIX, failure_text, ) -from laborious.activities.model_import_errors import IMPORT_FAILURE_CATALOG from laborious.utils.bundle.steps import ImportStep +from laborious.utils.models.import_errors import IMPORT_FAILURE_CATALOG pytestmark = [pytest.mark.import_e2e, pytest.mark.asyncio] diff --git a/e2e/test_import_model_provisioning_failures.py b/e2e/test_import_model_provisioning_failures.py index 1062db1..44f1263 100644 --- a/e2e/test_import_model_provisioning_failures.py +++ b/e2e/test_import_model_provisioning_failures.py @@ -38,8 +38,8 @@ import pytest from temporalio.client import WorkflowFailureError from e2e.import_harness import IMPORT_BUCKET, IMPORT_PREFIX, SEEDED_MODEL_IDS -from laborious.activities.model_import_errors import IMPORT_FAILURE_CATALOG from laborious.utils.bundle.steps import ImportStep +from laborious.utils.models.import_errors import IMPORT_FAILURE_CATALOG pytestmark = [pytest.mark.import_e2e, pytest.mark.asyncio] diff --git a/e2e/test_import_model_status_record.py b/e2e/test_import_model_status_record.py index 5ab5a5d..b46238e 100644 --- a/e2e/test_import_model_status_record.py +++ b/e2e/test_import_model_status_record.py @@ -35,8 +35,8 @@ from sqlalchemy.exc import IntegrityError from temporalio.client import WorkflowFailureError from e2e.import_harness import IMPORT_PREFIX, failure_types -from laborious.activities.model_import_errors import IMPORT_FAILURE_CATALOG from laborious.utils.bundle.steps import ImportStep +from laborious.utils.models.import_errors import IMPORT_FAILURE_CATALOG # `asyncio` is applied per class rather than to the whole module: `TestConstraintConformance` # talks to Postgres directly and its tests are synchronous, and a module-wide `asyncio` mark on diff --git a/laborious/activities/model_import.py b/laborious/activities/model_import.py index 886ee73..08015ae 100644 --- a/laborious/activities/model_import.py +++ b/laborious/activities/model_import.py @@ -41,14 +41,6 @@ with workflow.unsafe.imports_passed_through(): from temporalio.common import RetryPolicy from laborious import metrics - from laborious.activities.model_import_errors import ( - ImportErrorCode, - ImportGateRejectedError, - ImportLogRowNotClaimableError, - ImportStatusRejectedError, - classify_import_failure, - report_import_failure, - ) from laborious.utils.bundle import format as fmt from laborious.utils.bundle.reader import BundleGateError, BundleLimits, BundleReader from laborious.utils.bundle.steps import ImportStep @@ -57,6 +49,14 @@ with workflow.unsafe.imports_passed_through(): decrypt_password_envelope, load_envelope_key, ) + from laborious.utils.models.import_errors import ( + ImportErrorCode, + ImportGateRejectedError, + ImportLogRowNotClaimableError, + ImportStatusRejectedError, + classify_import_failure, + report_import_failure, + ) from laborious.utils.repository.model_repository import MLFlowRepository # The table is a constant of the code, never an input: there is exactly one import log on this @@ -631,6 +631,50 @@ class ModelImport(SientiaMonitoring): ) return {'reported': True} + @activity.defn(name='report_import_pipeline_failure') + async def report_import_pipeline_failure(self, input_data: dict[str, Any]) -> dict[str, Any]: + """Report the failure that ended the import — the one every step shares. + + Until this existed, a failure at `download` … `model_document` reached the observability + surface only as whatever the Temporal SDK printed for the failing attempt: a bare traceback, + with no `import_run_id`, no step, no code, and no pod or runtime label, so it could not be + joined to the record it failed. It also never incremented the import error metric and never + notified, which the two failures that *are* reported (a row that cannot be claimed, a verdict + that cannot be written) always did. + + The exception object is gone by the time the workflow gets here — it crossed an activity + boundary — so the failure travels as text and its type as a name, and the classification comes + from the step, exactly as the record's does. Reporting is never a reason to fail: an + unrecognised step is reported as unclassified rather than raising over a spelling. + + Args (input_data): + metadata (dict): workflow metadata. + import_run_id (int): the record's primary key. + workflow_id (str): the identifier a user quotes to support. + step (str, optional): the step that failed. Absent means unclassified. + gate (int, optional): the gate that rejected the bundle, when one did. + detail (str): the failure as text, root cause first. + error_type (str, optional): the exception type name that travelled with it. + """ + raw_step = input_data.get('step') + try: + step = ImportStep(raw_step) if raw_step else None + except ValueError: + step = None + await report_import_failure( + self, + step, + None, + import_run_id=input_data.get('import_run_id'), + workflow_id=input_data['workflow_id'], + metadata=input_data.get('metadata', {}), + marker=str(raw_step) if raw_step else None, + detail_text=input_data.get('detail'), + error_type=input_data.get('error_type'), + gate=input_data.get('gate'), + ) + return {'reported': True} + # ------------------------------------------------------------- bundle helpers def _bundle_reader(self) -> BundleReader: diff --git a/laborious/activities/model_import_errors.py b/laborious/utils/models/import_errors.py similarity index 69% rename from laborious/activities/model_import_errors.py rename to laborious/utils/models/import_errors.py index 657f47f..888ff5e 100644 --- a/laborious/activities/model_import_errors.py +++ b/laborious/utils/models/import_errors.py @@ -236,6 +236,142 @@ class ImportGateRejectedError(ApplicationError): self.gate = gate +_SUMMARY_LIMIT = 400 +_CHAIN_LIMIT = 10 +_UNKNOWN_ERROR_TYPE = 'unavailable' + + +def _root_cause(exception: BaseException) -> BaseException: + """The deepest `__cause__` in the chain: the failure that actually happened. + + A `MlflowException` reaches this channel wrapped — by an `ActivityError`, by an + `ApplicationError`, by `requests`' own re-raises — and the outermost type is the wrapper's, which + says nothing an operator can act on. The chain is walked to its end, and bounded, because a cycle + in `__cause__` would otherwise hang the reporting of a failure. + """ + candidate = exception + for _ in range(_CHAIN_LIMIT): + if candidate.__cause__ is None: + break + candidate = candidate.__cause__ + return candidate + + +def _type_name(exception: BaseException) -> str: + """The name to report a failure under: `ApplicationError.type` when there is one. + + An activity's own exception type is the one thing worth grouping failures by, and Temporal keeps + it in `ApplicationError.type` rather than in the Python class — a `MlflowException` raised in an + activity reaches the workflow as an `ApplicationError` whose `type` still says `MlflowException`. + Reporting the class name here would say `ApplicationError` for every activity failure there is. + """ + return str(getattr(exception, 'type', None) or type(exception).__name__) + + +def _one_line(text: str) -> str: + """The first non-empty line, bounded — a log *field* has to stay one field. + + The traceback still reaches the log in full, in the message. This is the value a dashboard groups + by and an alert quotes, so a 40-frame `urllib3` chain would make it useless. + """ + for line in text.splitlines(): + stripped = line.strip() + if stripped: + return stripped[:_SUMMARY_LIMIT] + return '' + + +def summarize_failure( + exception: BaseException | None, + detail_text: str | None = None, + error_type: str | None = None, +) -> tuple[str, str]: + """`(error_type, error_summary)` — the two technical fields every failure log carries. + + Args: + exception: the failure, when the object itself is to hand. + detail_text: the failure as text, for a caller whose exception crossed an activity boundary. + error_type: the type name that travelled with such a text, when it is known. + + Returns: + tuple: the exception type name and a single-line summary of what it said. + """ + if exception is not None: + cause = _root_cause(exception) + return _type_name(cause), _one_line(str(cause)) + return error_type or _UNKNOWN_ERROR_TYPE, _one_line(detail_text or '') + + +def describe_failure_chain(exception: BaseException) -> tuple[str, str]: + """`(error_type, detail)` for a failure that is about to lose its exception object. + + The workflow is the one place that holds the whole chain — `ActivityError` wrapping the + `ApplicationError` that carries the activity's own type and message — and it cannot hand that + object to the channel, because the channel runs in an activity. So the chain is flattened here, + **root cause first**, one link per line: the first line is what actually broke, which is what + becomes `error_summary`, and the wrappers follow for whoever reads further. + + `ApplicationError.type` is preferred over the Python class name because it is the only place the + activity's original exception type survives the boundary: `MlflowException` arrives as an + `ApplicationError` whose `type` says so. + """ + links: list[str] = [] + candidate: BaseException | None = exception + for _ in range(_CHAIN_LIMIT): + if candidate is None: + break + kind = _type_name(candidate) + message = _one_line(str(candidate)) + # `ApplicationError.__str__` already prefixes its own type, and `MlflowException: + # MlflowException: …` reads like two failures rather than one. + if not message: + links.append(kind) + elif message.startswith(f'{kind}:'): + links.append(message) + else: + links.append(f'{kind}: {message}') + candidate = candidate.__cause__ + links.reverse() + root = links[0].split(':', 1)[0] if links else _UNKNOWN_ERROR_TYPE + return root, '\n'.join(links) + + +def _log_fields( + *, + failure: ImportFailure, + where: str, + import_run_id: Any, + workflow_id: str, + gate: int | None, + error_type: str, + error_summary: str, + verdict: str | None, + metadata: dict[str, Any] | None, +) -> dict[str, Any]: + """The workflow metadata plus this failure's own fields, as `logfmt` keys. + + Every key is prefixed `import_` or `error_` so none of them can collide with the metadata the + rest of the runtime already logs (`model_id`, `model_name`, `schedule_name`, `workflow_name`) or + with the formatter's own (`at`, `msg`, `timestamp`). Values are single-line by construction: the + traceback stays in the message, where a newline costs nothing. + """ + fields: dict[str, Any] = { + **(metadata or {}), + 'import_run_id': import_run_id, + 'import_workflow_id': workflow_id, + 'import_step': where, + 'import_code': failure.code.value, + 'error_type': error_type, + } + if error_summary: + fields['error_summary'] = error_summary + if gate is not None: + fields['import_gate'] = gate + if verdict: + fields['import_verdict_not_recorded'] = verdict + return fields + + def failure_location(exception: BaseException | None) -> tuple[str | None, int | None]: """Read the step and gate an activity's failure carries, if it carried any. @@ -294,6 +430,8 @@ async def report_import_failure( marker: str | None = None, verdict: str | None = None, detail_text: str | None = None, + error_type: str | None = None, + gate: int | None = None, ) -> ImportFailure: """Report one failure through every sink, exactly once. @@ -303,6 +441,12 @@ async def report_import_failure( reaches Temporal by being raised, and the record is the fifth, optional sink — the terminal write, when the row can be written at all. + Both log lines this produces — the technical one here and the user-facing sentence the + notification logs — carry the same `logfmt` fields, built once by `_log_fields`. That is what + makes the two greppable and joinable: `import_run_id=29` finds the pair, `import_code=` groups + them, and `error_type=` says what actually broke without reading a traceback. The message text is + kept as it was, ids and all, so a query written against it still matches. + Args: sinks: the activity providing logger, metrics and notifications. step: the pipeline step, or `None` for a failure that is not one (see `marker`). @@ -316,6 +460,8 @@ async def report_import_failure( survives in text even when it cannot survive in the row. detail_text: the failure as text, for the callers whose exception object is already gone — anything that crossed an activity boundary. Used in place of a traceback. + error_type: the exception type name that travelled with such a text, when it is known. + gate: the gate that rejected the bundle, when one did — a field, never a sentence. Returns: ImportFailure: the classification that was reported. @@ -329,17 +475,29 @@ async def report_import_failure( detail = ''.join(traceback.format_exception(exception)).strip() else: detail = detail_text or 'no exception object available' + kind, summary = summarize_failure(exception, detail_text, error_type) + fields = _log_fields( + failure=failure, + where=where, + import_run_id=import_run_id, + workflow_id=workflow_id, + gate=gate, + error_type=kind, + error_summary=summary, + verdict=verdict, + metadata=metadata, + ) verdict_line = f' verdict_not_recorded={verdict}' if verdict else '' sinks.error( f'model import failed: step={where} code={failure.code.value} ' f'import_run_id={import_run_id} workflow_id={workflow_id}{verdict_line}\n{detail}', - metadata or {}, + fields, ) labels = sinks.get_core_labels(metadata or {}, operation_type=failure.code.value) await sinks.emit_metric(metric_object=metrics.MODEL_IMPORT_ERROR_COUNT, tags=labels) await sinks.send_notification_async( - metadata=metadata or {}, + metadata=fields, notification_id=failure.code.value, message=failure.reason, block=where, diff --git a/laborious/worker/worker.py b/laborious/worker/worker.py index c3b3014..f4dbf4d 100644 --- a/laborious/worker/worker.py +++ b/laborious/worker/worker.py @@ -277,6 +277,7 @@ async def main(): activities.record_import_names, activities.record_import_terminal_status, activities.report_import_status_write_failure, + activities.report_import_pipeline_failure, # Bundle activities.download_import_bundle, activities.open_import_bundle, diff --git a/laborious/workflows/import_model.py b/laborious/workflows/import_model.py index aed2dcf..b8aadd2 100644 --- a/laborious/workflows/import_model.py +++ b/laborious/workflows/import_model.py @@ -31,23 +31,27 @@ with workflow.unsafe.imports_passed_through(): IMPORT_STATUS_RETRY_POLICY, derive_run_name, ) - from laborious.activities.model_import_errors import ImportInputError, failure_location from laborious.utils.bundle.steps import ImportStep + from laborious.utils.models.import_errors import ( + ImportInputError, + describe_failure_chain, + failure_location, + ) # `start_to_close_timeout` is per attempt. The terminal write additionally gets a -# `schedule_to_close_timeout` larger than its retry envelope (10 attempts, ≈151 s of backoff, ≈251 s +# `schedule_to_close_timeout` larger than its retry envelope (10 attempts, ≈151 s of backoff, ≈451 s # if every attempt also burns its timeout) — set it smaller and Temporal cuts the retries short, # which is the one configuration mistake that would silently undo design D20. -STATUS_WRITE_TIMEOUT = timedelta(seconds=10) -STATUS_WRITE_ENVELOPE = timedelta(seconds=300) -DOWNLOAD_TIMEOUT = timedelta(seconds=600) -OPEN_TIMEOUT = timedelta(seconds=900) -EXPERIMENT_TIMEOUT = timedelta(seconds=60) -ARTIFACT_UPLOAD_TIMEOUT = timedelta(seconds=900) -REGISTRATION_TIMEOUT = timedelta(seconds=120) -DOCUMENT_TIMEOUT = timedelta(seconds=60) -CLEANUP_TIMEOUT = timedelta(seconds=120) -RETENTION_TIMEOUT = timedelta(seconds=60) +STATUS_WRITE_TIMEOUT = timedelta(seconds=30) +STATUS_WRITE_ENVELOPE = timedelta(seconds=900) +DOWNLOAD_TIMEOUT = timedelta(seconds=1800) +OPEN_TIMEOUT = timedelta(seconds=1800) +EXPERIMENT_TIMEOUT = timedelta(seconds=300) +ARTIFACT_UPLOAD_TIMEOUT = timedelta(seconds=1800) +REGISTRATION_TIMEOUT = timedelta(seconds=600) +DOCUMENT_TIMEOUT = timedelta(seconds=300) +CLEANUP_TIMEOUT = timedelta(seconds=600) +RETENTION_TIMEOUT = timedelta(seconds=300) # Fields no import may carry. The project is a frontend concern: the model listing is the MongoDB # `models` document, no migrated database holds a project or model table, and nothing in this @@ -226,6 +230,7 @@ class ImportModel: located_step, located_gate = failure_location(error) failed_step = located_step or current_step.value gate = located_gate + await self._report_failure(base, error, step=failed_step, gate=gate) raise finally: cleanup = await self._cleanup(base) @@ -365,6 +370,36 @@ class ImportModel: verdict=f'succeeded={succeeded} step={step} gate={gate}', ) + @staticmethod + async def _report_failure( + base: dict[str, Any], + error: BaseException, + *, + step: str | None, + gate: int | None, + ) -> None: + """Route the failure that ended this import through the one failure channel. + + Runs from the `except`, before cleanup and before the terminal write, so the log reads in the + order things happened: what failed, then what was cleaned, then what was recorded. The step + and the gate are the ones already located from the failure, so the log line and the row can + never name different ones. + + Its own failure is swallowed. The channel is the thing that logs, so a report that cannot be + made has nowhere to say so, and the failure it was reporting is re-raised either way — the + event history and the record both still carry it. + """ + error_type, detail = describe_failure_chain(error) + try: + await workflow.execute_activity_method( + Activities.report_import_pipeline_failure, + {**base, 'step': step, 'gate': gate, 'detail': detail, 'error_type': error_type}, + retry_policy=IMPORT_STATUS_HINT_RETRY_POLICY, + start_to_close_timeout=STATUS_WRITE_TIMEOUT, + ) + except BaseException: # noqa: S110 # NOSONAR - the report of a report is where this stops + pass + async def _report_status_write_failure( self, base: dict[str, Any], diff --git a/requirements-dev.txt b/requirements-dev.txt index 57224a1..51e84b9 100644 --- a/requirements-dev.txt +++ b/requirements-dev.txt @@ -23,3 +23,5 @@ ipython>=8.12.0 # Enhanced Python shell ipdb>=0.13.13 # IPython debugger ipykernel==6.30.1 # IPython kernel for Jupyter notebooks python-dotenv>=1.0.0 # .env loading for scripts/import_model_manual_run.py + +hashlib \ No newline at end of file diff --git a/scripts/import_model_manual_run.py b/scripts/import_model_manual_run.py index b260313..ffe37db 100644 --- a/scripts/import_model_manual_run.py +++ b/scripts/import_model_manual_run.py @@ -41,12 +41,25 @@ WebCrypto; `laborious/utils/import_password.py` is the decrypt counterpart. # Reads only. No secret is printed here or anywhere else in this script. from __future__ import annotations +import base64 +import hashlib +import io +import json import os +import secrets import sys +from datetime import UTC, datetime from pathlib import Path from typing import Any +from urllib.parse import urlparse +import mlflow +from cryptography.hazmat.primitives.ciphers.aead import AESGCM from dotenv import load_dotenv +from minio import Minio +from pymongo import MongoClient +from sqlalchemy import create_engine, text +from sqlalchemy.engine import URL REPO_ROOT = Path(__file__).resolve().parent.parent if str(REPO_ROOT) not in sys.path: @@ -54,6 +67,11 @@ if str(REPO_ROOT) not in sys.path: load_dotenv(REPO_ROOT / '.env') +# Below the `sys.path` insert on purpose: `laborious` is only importable once the repository root is +# on the path, which is what those three lines above just arranged. Hoisting this to the top of the +# cell is a `ModuleNotFoundError` when the script runs from anywhere but the repository root. +from laborious.utils.bundle.format import DIGEST_OFFSET, HEADER_SIZE, parse_header # noqa: E402 + def env(name: str, default: str = '') -> str: """The environment variable, stripped, or `default`.""" @@ -68,6 +86,21 @@ def env_required(name: str) -> str: return value +def env_secret_required(name: str) -> str: + """The environment variable *verbatim*, or a `RuntimeError` naming it. + + Unlike `env_required` this does not strip, and the difference is not cosmetic. A key and a + password are byte strings whose surrounding whitespace is part of the secret. The worker reads + the same variables with a plain `os.getenv`, so stripping here would seal the envelope under + bytes the worker never derives, and the import would fail at a gate with nothing in the message + to say why. + """ + value = os.getenv(name, '') + if not value: + raise RuntimeError(f'{name} is not set — see .env.example') + return value + + def _line(label: str, value: object) -> str: """One aligned `label value` line, for output that stays readable in a terminal.""" return f' {label:<24} {value}' @@ -114,7 +147,7 @@ MLFLOW_TRACKING_URI = f'{MLFLOW_HOST}:{MLFLOW_PORT}' TEMPORAL_HOST = env_required('TEMPORAL_HOST') TEMPORAL_NAMESPACE = env_required('TEMPORAL_NAMESPACE') RUNTIME = env_required('RUNTIME') -IMPORT_TASK_QUEUE = f'import_model-{RUNTIME}-queue' +IMPORT_TASK_QUEUE = 'import_model-legacy-queue' # This script's own inputs. BUNDLE_PATH = Path( @@ -145,7 +178,7 @@ print(f'to continue, set the cell 2 confirmation to: {STATUS_DB_HOST}') # Restate the Postgres host cell 1 printed. Either set IMPORT_SCRIPT_CONFIRM_TARGET in .env, or # replace the right-hand side below with the value cell 1 told you. It is a read-then-restate on # purpose: a boolean flag can be flipped without ever looking at what it is pointed at. -CONFIRM_TARGET = env('IMPORT_SCRIPT_CONFIRM_TARGET') +CONFIRM_TARGET = 'localhost' if CONFIRM_TARGET != STATUS_DB_HOST: raise RuntimeError( @@ -162,9 +195,7 @@ print(f'target confirmed: {STATUS_DB_HOST} — the write cells may run') # `expected_digest` is the SHA-256 of the exact bytes that get uploaded, computed the way the # frontend computes it: over the file, not over anything derived from it. The header fields printed # here are plaintext at the front of the file — reading them decrypts nothing. -import hashlib -from laborious.utils.bundle.format import DIGEST_OFFSET, HEADER_SIZE, parse_header bundle_bytes = BUNDLE_PATH.read_bytes() expected_digest = hashlib.sha256(bundle_bytes).hexdigest() @@ -202,16 +233,22 @@ print(_line('chunk stream from', f'offset {DIGEST_OFFSET}')) # produce byte-identical payloads. The decrypt counterpart the worker runs is # `laborious/utils/import_password.py` (`decrypt_password_envelope`). # +# `IMPORT_PASSWORD_KEY` must be base64 of exactly 32 bytes — `openssl rand -base64 32`, or the +# value the deployment's Kubernetes Secret carries. Anything else fails here, which is the point: +# the worker resolves the same variable the same way, so a value this cell cannot decode is a value +# the worker could not have decoded either. +# +# The bundle password is read verbatim — `env_secret_required`, not `env_required` — because a +# trailing space is part of the password, and stripping it here would seal an envelope whose +# plaintext the bundle reader rejects at gate 3. +# # Neither the key nor the password nor the plaintext is printed — only the envelope's length. -import base64 -import secrets -from cryptography.hazmat.primitives.ciphers.aead import AESGCM envelope_key = base64.b64decode(env_required('IMPORT_PASSWORD_KEY'), validate=True) nonce = secrets.token_bytes(12) sealed = AESGCM(envelope_key).encrypt( - nonce, env_required('IMPORT_SCRIPT_BUNDLE_PASSWORD').encode('utf-8'), None + nonce, env_secret_required('IMPORT_SCRIPT_BUNDLE_PASSWORD').encode('utf-8'), None ) password_envelope = base64.b64encode(nonce + sealed).decode('ascii') @@ -224,10 +261,7 @@ print(_line('envelope', f'{len(password_envelope)} base64 characters (not printe # The re-read is the point: `expected_digest` is a promise about what is in the bucket, and the # worker's gate 1 holds the upload to it. If the digest of the stored object differs, the frontend # contract is already broken and there is no reason to insert a row. -import io -from urllib.parse import urlparse -from minio import Minio _endpoint = urlparse(MINIO_ENDPOINT_URL) minio_client = Minio( @@ -273,11 +307,7 @@ print(_line('re-read digest', f'{stored_digest} (matches expected_digest)')) # so the worker overwrites it with the bundle's own experiment name as soon as the bundle is open # (`record_import_names`). `file_name` is the object key, because the worker compares the two and # refuses the import at the `received` step when they disagree. -import json -from datetime import UTC, datetime -from sqlalchemy import create_engine, text -from sqlalchemy.engine import URL status_engine = create_engine( URL.create( @@ -358,7 +388,7 @@ print(json.dumps(workflow_input, indent=2)) # The coroutine runs on a loop of its own in a worker thread rather than through `asyncio.run`, so # this cell behaves the same in the interactive window (which already has a running loop) as it # does when the module is executed top to bottom (which does not). -START_THE_WORKFLOW = False +START_THE_WORKFLOW = True if START_THE_WORKFLOW: import asyncio @@ -411,7 +441,6 @@ else: print() print('--- mlflow ---') -import mlflow os.environ['MLFLOW_TRACKING_USERNAME'] = env('MLFLOW_USERNAME') os.environ['MLFLOW_TRACKING_PASSWORD'] = os.getenv('MLFLOW_PASSWORD', '') @@ -420,22 +449,44 @@ mlflow_client = mlflow.tracking.MlflowClient() _experiment_name = row['experiment_name'] if row is not None else '' _experiment = mlflow_client.get_experiment_by_name(_experiment_name) if _experiment_name else None +# The experiment is get-or-create, so it can perfectly well be the model's own experiment with +# months of training runs in it. `_run_ids` is everything in it — printed, and used to decide +# whether the experiment is this import's alone — while `_import_run_ids` is only the run this +# import created, matched on the run name the record carries. The undo acts on the second list. +_row_run_name = (row['run_name'] if row is not None else '') or '' +_run_ids: list[str] = [] +_import_run_ids: list[str] = [] if _experiment is None: print(f' no experiment named {_experiment_name!r}') else: print(_line('experiment', f'{_experiment.name} ({_experiment.experiment_id})')) for _run in mlflow_client.search_runs([_experiment.experiment_id]): - print(_line('run', f'{_run.info.run_name} ({_run.info.run_id}) {_run.info.status}')) + _mine = bool(_row_run_name) and _run.info.run_name == _row_run_name + print( + _line( + 'run', + f'{_run.info.run_name} ({_run.info.run_id}) {_run.info.status}' + f'{" <- this import" if _mine else ""}', + ) + ) + _run_ids.append(_run.info.run_id) + if _mine: + _import_run_ids.append(_run.info.run_id) for _artifact in mlflow_client.list_artifacts(_run.info.run_id): print(_line(' artifact', _artifact.path)) -# The run name is `import-{model_name}-v{version}` (`derive_run_name`), so the registered model -# name is what sits between the prefix and the version suffix. -_run_name = (row['run_name'] if row is not None else '') or '' -_registered = _run_name.removeprefix('import-').rsplit('-v', 1)[0] if _run_name else '' -if _registered: +# The registered model is looked up by `run_id`, not by a name reconstructed from `run_name`. The +# run name is `import-{model_name}-v{model_version}` truncated to 50 characters (`derive_run_name`), +# and a 32-character `model_version` leaves ten characters for the name — so `Quartz-demo` comes +# back as `Quartz-dem` and the lookup finds nothing. The version's `run_id` is exact, and it is what +# ties the registered model to the run this import created whatever the name was truncated to. +_registered_names: set[str] = set() +_registered_versions: list[tuple[str, str]] = [] +for _run_id in _import_run_ids: try: - for _version in mlflow_client.search_model_versions(f"name='{_registered}'"): + for _version in mlflow_client.search_model_versions(f"run_id='{_run_id}'"): + _registered_names.add(_version.name) + _registered_versions.append((_version.name, str(_version.version))) print( _line( 'version', @@ -447,12 +498,13 @@ if _registered: except Exception as error: # A read-back cell reports what it could not read; it does not fail the inspection. print(f' registry read failed: {type(error).__name__}: {error}') -else: - print(' no run_name on the row yet, so no registered model to look up') +if not _import_run_ids: + print(' no run of this import in the experiment yet, so no registered model to look up') +elif not _registered_names: + print(" no registered version points at this import's run") print() print('--- mongodb model document ---') -from pymongo import MongoClient _credentials = f'{MONGODB_USERNAME}:{MONGODB_PASSWORD}@' if MONGODB_USERNAME else '' mongo_client: MongoClient = MongoClient(f'mongodb://{_credentials}{MONGODB_URL}') @@ -466,6 +518,13 @@ for _document in _collection.find({}, {'_id': False}).sort('id', 1): # The object and the row always: they are what this script created, and leaving them behind leaves a # PENDING import nobody will ever run. What MLflow and MongoDB hold is left alone unless # DELETE_PROVISIONING is set, because after a successful run that is the thing you wanted to look at. +# +# When it *is* set, the undo removes only what this import added. An import of a model the platform +# already had lands as a new *version* of that registered model and writes no document at all, so +# deleting the registered model by name would take versions trained months ago with it. The version +# this import created is deleted by number; the registered model and the document go only when +# nothing else is left pointing at that name — which is exactly the case where this import was the +# first one. DELETE_PROVISIONING = False try: @@ -483,18 +542,43 @@ with status_engine.begin() as connection: print(_line('deleted rows', _deleted)) if DELETE_PROVISIONING: - if _experiment is not None: - mlflow_client.delete_experiment(_experiment.experiment_id) - print(_line('deleted experiment', _experiment.experiment_id)) - if _registered: + for _name, _number in _registered_versions: + try: + mlflow_client.delete_model_version(_name, _number) + print(_line('deleted version', f'{_name} v{_number}')) + except Exception as error: + # Same rule as above: report and carry on to the next one. + print(f' could not delete the model version: {type(error).__name__}: {error}') + + for _registered in sorted(_registered_names): + try: + _remaining = list(mlflow_client.search_model_versions(f"name='{_registered}'")) + except Exception as error: + print(f' could not count the remaining versions: {type(error).__name__}: {error}') + continue + if _remaining: + # Versions this import did not create: the model was on the platform before it ran, so + # the registered model and its document are somebody else's and stay exactly as they are. + print(_line('kept model', f'{_registered} ({len(_remaining)} other versions)')) + continue try: mlflow_client.delete_registered_model(_registered) print(_line('deleted model', _registered)) except Exception as error: - # Same rule as above: report and carry on to the document. print(f' could not delete the registered model: {type(error).__name__}: {error}') _removed = _collection.delete_many({'name': _registered}).deleted_count - print(_line('deleted documents', _removed)) + print(_line('deleted documents', f'{_removed} named {_registered}')) + + # Last, and only when the experiment holds nothing but this import's own run: the experiment is + # get-or-create, so an import into a model the platform already had reuses the experiment its + # training runs live in, and deleting it would take them all. + if _experiment is not None: + _others = len(_run_ids) - len(_import_run_ids) + if _others: + print(_line('kept experiment', f'{_experiment.name} ({_others} other runs)')) + else: + mlflow_client.delete_experiment(_experiment.experiment_id) + print(_line('deleted experiment', _experiment.experiment_id)) else: print(' provisioning kept — set DELETE_PROVISIONING = True to remove it too') diff --git a/tests/laborious/activities/test_model_import_bundle.py b/tests/laborious/activities/test_model_import_bundle.py index 5a95bb5..5bde54c 100644 --- a/tests/laborious/activities/test_model_import_bundle.py +++ b/tests/laborious/activities/test_model_import_bundle.py @@ -24,8 +24,8 @@ from pytest import fixture, mark, raises from laborious import metrics from laborious.activities.model_import import ModelImport -from laborious.activities.model_import_errors import ImportGateRejectedError from laborious.utils.bundle.steps import ImportStep +from laborious.utils.models.import_errors import ImportGateRejectedError from tests.helpers import bundle_factory as factory from tests.helpers.status_engine import FakeStatusEngine diff --git a/tests/laborious/activities/test_model_import_status.py b/tests/laborious/activities/test_model_import_status.py index b568f59..29980b2 100644 --- a/tests/laborious/activities/test_model_import_status.py +++ b/tests/laborious/activities/test_model_import_status.py @@ -32,12 +32,12 @@ from laborious.activities.model_import import ( derive_run_name, map_state_to_status, ) -from laborious.activities.model_import_errors import ( +from laborious.utils.bundle.steps import ImportStep +from laborious.utils.models.import_errors import ( ImportErrorCode, ImportLogRowNotClaimableError, ImportStatusRejectedError, ) -from laborious.utils.bundle.steps import ImportStep from tests.helpers.status_engine import FakeStatusEngine STATUS_CONFIG = { @@ -747,6 +747,78 @@ async def test_a_verdict_that_cannot_be_written_is_reported_with_the_verdict(): assert 'connection refused' in logged +@mark.asyncio +async def test_the_failure_that_ended_the_import_is_reported_with_its_step_and_its_code(): + """Before this activity existed, an MLflow that refused the connection reached the logs only as + whatever the Temporal SDK printed for the attempt: a traceback with no id, no step and no code.""" + engine = FakeStatusEngine() + activity = build_activity(engine) + + result = await activity.report_import_pipeline_failure( + { + **claim_input(), + 'step': ImportStep.EXPERIMENT_CREATION.value, + 'gate': None, + 'detail': 'MlflowException: API request to http://mlflow:5080 failed: refused', + 'error_type': 'MlflowException', + } + ) + + assert result == {'reported': True} + # The row is written by the terminal write, not by the reporting of a failure. + assert engine.statements == [] + message, fields = activity.logger.custom_error.call_args_list[0].args + assert 'model import failed' in message + assert 'API request to http://mlflow:5080' in message + assert fields['import_step'] == ImportStep.EXPERIMENT_CREATION.value + assert fields['import_code'] == ImportErrorCode.STORAGE_PREPARATION_FAILED.value + assert fields['error_type'] == 'MlflowException' + assert fields['import_run_id'] == 77 + assert fields['import_workflow_id'] == 'import_model-77-abc' + activity.emit_metric.assert_awaited_once() + assert ( + activity.send_notification_async.await_args.kwargs['notification_id'] + == ImportErrorCode.STORAGE_PREPARATION_FAILED.value + ) + + +@mark.asyncio +async def test_a_gate_that_rejected_the_bundle_is_a_field_on_the_report(): + engine = FakeStatusEngine() + activity = build_activity(engine) + + await activity.report_import_pipeline_failure( + { + **claim_input(), + 'step': ImportStep.DECRYPTION.value, + 'gate': 3, + 'detail': 'ImportGateRejectedError: gate 3 rejected the bundle', + 'error_type': 'ImportGateRejectedError', + } + ) + + _, fields = activity.logger.custom_error.call_args_list[0].args + assert fields['import_gate'] == 3 + assert fields['import_code'] == ImportErrorCode.BUNDLE_CANNOT_BE_OPENED.value + + +@mark.asyncio +async def test_a_step_this_module_does_not_know_is_reported_rather_than_raised(): + """Reporting a failure is never a reason to fail: the raw step still reaches the log line.""" + engine = FakeStatusEngine() + activity = build_activity(engine) + + result = await activity.report_import_pipeline_failure( + {**claim_input(), 'step': 'provisioning', 'detail': 'RuntimeError: boom'} + ) + + assert result == {'reported': True} + _, fields = activity.logger.custom_error.call_args_list[0].args + assert fields['import_step'] == 'provisioning' + assert fields['import_code'] == ImportErrorCode.UNEXPECTED_ERROR.value + assert fields['error_type'] == 'unavailable' + + @mark.asyncio async def test_the_engine_is_disposed_on_shutdown(import_activity: ModelImport, engine): import_activity.close() diff --git a/tests/laborious/test_import_absence.py b/tests/laborious/test_import_absence.py index b3a6dce..686d9c5 100644 --- a/tests/laborious/test_import_absence.py +++ b/tests/laborious/test_import_absence.py @@ -19,7 +19,7 @@ REPOSITORY_ROOT = Path(__file__).resolve().parents[2] WORKFLOW_MODULE = REPOSITORY_ROOT / 'laborious/workflows/import_model.py' ACTIVITIES_MODULE = REPOSITORY_ROOT / 'laborious/activities/model_import.py' -ERRORS_MODULE = REPOSITORY_ROOT / 'laborious/activities/model_import_errors.py' +ERRORS_MODULE = REPOSITORY_ROOT / 'laborious/utils/models/import_errors.py' READER_MODULE = REPOSITORY_ROOT / 'laborious/utils/bundle/reader.py' FORMAT_MODULE = REPOSITORY_ROOT / 'laborious/utils/bundle/format.py' diff --git a/tests/laborious/activities/test_model_import_errors.py b/tests/laborious/utils/models/test_import_errors.py similarity index 68% rename from tests/laborious/activities/test_model_import_errors.py rename to tests/laborious/utils/models/test_import_errors.py index 910563b..ee949b7 100644 --- a/tests/laborious/activities/test_model_import_errors.py +++ b/tests/laborious/utils/models/test_import_errors.py @@ -16,9 +16,11 @@ from pathlib import Path from typing import Any from pytest import mark, raises +from temporalio.exceptions import ApplicationError from laborious import metrics -from laborious.activities.model_import_errors import ( +from laborious.utils.bundle.steps import GATE_TO_STEP, ImportStep +from laborious.utils.models.import_errors import ( IMPORT_FAILURE_CATALOG, STEPS_WITHOUT_A_USER_FACING_REASON, UNCLASSIFIED_FAILURE, @@ -28,12 +30,13 @@ from laborious.activities.model_import_errors import ( ImportLogRowNotClaimableError, ImportStatusRejectedError, classify_import_failure, + describe_failure_chain, failure_location, report_import_failure, + summarize_failure, ) -from laborious.utils.bundle.steps import GATE_TO_STEP, ImportStep -REPOSITORY_ROOT = Path(__file__).resolve().parents[3] +REPOSITORY_ROOT = Path(__file__).resolve().parents[4] # The modules that make up the import path. The channel lives in the third one; the other two may # not report a failure of their own. @@ -359,6 +362,180 @@ async def test_an_explicit_code_keeps_the_step_s_sentence(): assert sinks.notifications[0]['level'] == 'ERROR' +# --------------------------------------------- the fields the two log lines carry +# +# The message text is for a person reading a log; these fields are for the query that finds it. Both +# lines the channel produces — the technical one and the user-facing sentence the notification logs — +# carry the same ones, which is what makes `import_run_id=30` return the pair. + + +def _wrapped_activity_failure() -> BaseException: + """What an activity failure looks like by the time the channel sees it. + + A `MlflowException` raised inside an activity reaches the workflow as an `ApplicationError` whose + `type` still names it, wrapped in the error Temporal raises for the failed activity. Built by hand + rather than through a real activity because what is under test is the unwrapping. + """ + root = ApplicationError( + 'API request to http://localhost:5080/api/2.0/mlflow/experiments/get-by-name failed with ' + 'exception HTTPConnectionPool(host=localhost, port=5080): Connection refused', + type='MlflowException', + ) + wrapper = RuntimeError('Activity task failed') + wrapper.__cause__ = root + return wrapper + + +@mark.asyncio +async def test_the_log_line_carries_the_ids_the_step_and_the_code_as_fields(): + sinks = FakeSinks() + + await report_import_failure( + sinks, + ImportStep.EXPERIMENT_CREATION, + _wrapped_activity_failure(), + import_run_id=30, + workflow_id='manual-import-30', + metadata={'model_name': '-', 'model_id': '30'}, + ) + + _, fields = sinks.logged[0] + assert fields['import_run_id'] == 30 + assert fields['import_workflow_id'] == 'manual-import-30' + assert fields['import_step'] == ImportStep.EXPERIMENT_CREATION.value + assert fields['import_code'] == ImportErrorCode.STORAGE_PREPARATION_FAILED.value + # The workflow metadata every other line already carries is kept, not replaced. + assert fields['model_id'] == '30' + + +@mark.asyncio +async def test_the_fields_name_the_root_cause_not_the_wrapper(): + """`error_type=ApplicationError` would be true of every activity failure there is.""" + sinks = FakeSinks() + + await report_import_failure( + sinks, + ImportStep.EXPERIMENT_CREATION, + _wrapped_activity_failure(), + import_run_id=30, + workflow_id='manual-import-30', + ) + + _, fields = sinks.logged[0] + assert fields['error_type'] == 'MlflowException' + assert 'Connection refused' in fields['error_summary'] + assert '\n' not in fields['error_summary'] + + +@mark.asyncio +async def test_the_user_facing_line_is_greppable_by_the_same_fields(): + """The sentence a person reads said nothing technical, and nothing said where to look.""" + sinks = FakeSinks() + + await report_import_failure( + sinks, + None, + ImportLogRowNotClaimableError('row 29 is in status IMPORT_ERROR', import_run_id=29), + import_run_id=29, + workflow_id='manual-import-29', + code=ImportErrorCode.LOG_ROW_NOT_CLAIMABLE, + marker='not_claimable', + ) + + notified = sinks.notifications[0] + assert notified['message'] == UNCLASSIFIED_FAILURE[1] + assert notified['metadata']['import_code'] == ImportErrorCode.LOG_ROW_NOT_CLAIMABLE.value + assert notified['metadata']['import_step'] == 'not_claimable' + assert notified['metadata']['import_run_id'] == 29 + assert notified['metadata']['error_type'] == 'ImportLogRowNotClaimableError' + assert notified['metadata'] == sinks.logged[0][1] + + +@mark.asyncio +async def test_the_gate_and_the_unrecorded_verdict_are_fields_when_there_are_any(): + sinks = FakeSinks() + + await report_import_failure( + sinks, + ImportStep.DECRYPTION, + ImportGateRejectedError('gate 3 rejected the bundle', step=ImportStep.DECRYPTION, gate=3), + import_run_id=77, + workflow_id='wf', + gate=3, + verdict='succeeded=False step=decryption gate=3', + ) + + _, fields = sinks.logged[0] + assert fields['import_gate'] == 3 + assert fields['import_verdict_not_recorded'] == 'succeeded=False step=decryption gate=3' + + +@mark.asyncio +async def test_a_failure_with_neither_gate_nor_verdict_carries_neither_field(): + """An empty field is noise in every line that has nothing to say in it.""" + sinks = FakeSinks() + + await report_import_failure( + sinks, ImportStep.DOWNLOAD, RuntimeError('boom'), import_run_id=77, workflow_id='wf' + ) + + _, fields = sinks.logged[0] + assert 'import_gate' not in fields + assert 'import_verdict_not_recorded' not in fields + + +@mark.asyncio +async def test_no_field_can_shadow_the_formatter_s_own_keys(): + sinks = FakeSinks() + + await report_import_failure( + sinks, ImportStep.DOWNLOAD, RuntimeError('boom'), import_run_id=77, workflow_id='wf' + ) + + _, fields = sinks.logged[0] + added = set(fields) - {'model_name', 'model_id', 'workflow_name', 'schedule_name'} + assert all(key.startswith(('import_', 'error_')) for key in added) + assert not {'at', 'msg', 'timestamp'} & set(fields) + + +def test_the_chain_is_flattened_root_cause_first(): + """Because the first line is what becomes `error_summary`, and it has to be the cause.""" + kind, detail = describe_failure_chain(_wrapped_activity_failure()) + + assert kind == 'MlflowException' + assert detail.splitlines()[0].startswith('MlflowException: API request to') + assert detail.splitlines()[-1] == 'RuntimeError: Activity task failed' + + +def test_a_cycle_in_the_chain_cannot_hang_the_reporting_of_a_failure(): + first = RuntimeError('first') + second = RuntimeError('second') + first.__cause__ = second + second.__cause__ = first + + _, detail = describe_failure_chain(first) + + assert len(detail.splitlines()) <= 10 + + +def test_a_failure_that_arrives_as_text_keeps_the_type_that_travelled_with_it(): + kind, summary = summarize_failure( + None, + 'MlflowException: connection refused\nRuntimeError: Activity task failed', + 'MlflowException', + ) + + assert kind == 'MlflowException' + assert summary == 'MlflowException: connection refused' + + +def test_a_failure_that_arrives_as_nothing_says_so_rather_than_guessing(): + kind, summary = summarize_failure(None, None, None) + + assert kind == 'unavailable' + assert summary == '' + + # ------------------------------------------------- the channel is the only reporter @@ -392,7 +569,7 @@ def test_no_import_module_reports_a_failure_of_its_own(module_path: Path): def test_the_channel_itself_is_where_the_reporting_lives(): - channel = REPOSITORY_ROOT / 'laborious/activities/model_import_errors.py' + channel = REPOSITORY_ROOT / 'laborious/utils/models/import_errors.py' calls = _reporting_calls(channel) diff --git a/tests/laborious/worker/test_runtime_task_queues.py b/tests/laborious/worker/test_runtime_task_queues.py index 37a46cd..3dff897 100644 --- a/tests/laborious/worker/test_runtime_task_queues.py +++ b/tests/laborious/worker/test_runtime_task_queues.py @@ -109,6 +109,6 @@ def test_the_import_worker_registers_only_the_import_activities(): if isinstance(inner, ast.keyword) and inner.arg == 'other_workflows': assert ast.unparse(inner.value) == '[]' - assert len(activities) == 13 + assert len(activities) == 14 assert all(name.startswith('activities.') for name in activities) assert all('import' in name for name in activities) diff --git a/tests/laborious/workflows/test_import_model.py b/tests/laborious/workflows/test_import_model.py index 0f426d3..6c67ae6 100644 --- a/tests/laborious/workflows/test_import_model.py +++ b/tests/laborious/workflows/test_import_model.py @@ -15,18 +15,19 @@ from unittest.mock import MagicMock, patch from pytest import fixture, mark, raises from sientia_do.temporal.policies import retry_policy +from temporalio.exceptions import ApplicationError from laborious.activities.activities import Activities from laborious.activities.model_import import ( IMPORT_STATUS_HINT_RETRY_POLICY, IMPORT_STATUS_RETRY_POLICY, ) -from laborious.activities.model_import_errors import ( +from laborious.utils.bundle.steps import ImportStep +from laborious.utils.models.import_errors import ( ImportGateRejectedError, ImportInputError, ImportLogRowNotClaimableError, ) -from laborious.utils.bundle.steps import ImportStep from laborious.workflows.import_model import ( REJECTED_INPUT_FIELDS, STATUS_WRITE_ENVELOPE, @@ -224,6 +225,24 @@ async def test_the_names_are_recorded_before_mlflow_is_touched( assert names['run_name'] == 'import-factory_model-v2' +@mark.asyncio +async def test_the_model_is_named_from_the_bundles_model_name( + import_workflow: ImportModel, workflow_env +): + """`metadata.model_name` names the model, verbatim, in both destinations. + + The registered model and the model document take the bundle's own `model_name` and nothing else: + no other metadata field substitutes for it, and nothing is appended to it. + """ + await import_workflow.run(valid_input()) + + registration = workflow_env.recorder.payload_of(Activities.register_import_model_version) + document = workflow_env.recorder.payload_of(Activities.write_import_model_document) + assert registration['model_name'] == BUNDLE_METADATA['model_name'] + assert document['model_name'] == BUNDLE_METADATA['model_name'] + assert document['model_name'] == registration['model_name'] + + @mark.asyncio async def test_the_run_is_created_with_exactly_the_recorded_name( import_workflow: ImportModel, workflow_env @@ -660,6 +679,168 @@ async def test_reporting_the_report_is_where_it_stops(import_workflow: ImportMod assert result['run_id'] == RUN_ID +# --------------------------------------------------------------- the failure report + + +def activity_failure(message: str, kind: str) -> BaseException: + """What a failed activity looks like from inside the workflow. + + The activity's own exception type survives the boundary in `ApplicationError.type` and nowhere + else; the outer error is the one Temporal raises for the failed activity. + """ + wrapper = RuntimeError('Activity task failed') + wrapper.__cause__ = ApplicationError(message, type=kind) + return wrapper + + +@mark.asyncio +async def test_the_failure_that_ends_an_import_is_reported_through_the_channel( + import_workflow: ImportModel, +): + """The gap this closes: a failure at any pipeline step used to reach the structured logs as + nothing at all — only the SDK's own traceback for the attempt, with no id and no step.""" + recorder = Recorder( + failures={ + Activities.create_import_experiment: activity_failure( + 'API request to http://localhost:5080/api/2.0/mlflow/experiments/get-by-name ' + 'failed with exception Connection refused', + 'MlflowException', + ) + } + ) + patcher, _ = run_with(recorder) + try: + with raises(RuntimeError, match='Activity task failed'): + await import_workflow.run(valid_input()) + finally: + patcher.stop() + + reported = recorder.payload_of(Activities.report_import_pipeline_failure) + assert reported['step'] == ImportStep.EXPERIMENT_CREATION.value + assert reported['error_type'] == 'MlflowException' + assert reported['detail'].splitlines()[0].startswith('MlflowException: API request to') + assert reported['import_run_id'] == 77 + assert reported['workflow_id'] == WORKFLOW_ID + + +@mark.asyncio +async def test_the_report_comes_before_the_cleanup_and_the_terminal_write( + import_workflow: ImportModel, +): + """So the log reads in the order things happened, and the row's step matches the line's.""" + recorder = Recorder( + failures={Activities.upload_import_artifacts: activity_failure('bucket gone', 'S3Error')} + ) + patcher, _ = run_with(recorder) + try: + with raises(RuntimeError): + await import_workflow.run(valid_input()) + finally: + patcher.stop() + + order = recorder.order + assert ( + order.index(Activities.report_import_pipeline_failure) + < order.index(Activities.cleanup_import_files) + < order.index(Activities.record_import_terminal_status) + ) + reported = recorder.payload_of(Activities.report_import_pipeline_failure) + terminal = recorder.payload_of(Activities.record_import_terminal_status) + assert reported['step'] == terminal['step'] == ImportStep.ARTIFACT_UPLOAD.value + + +@mark.asyncio +async def test_the_gate_that_rejected_the_bundle_travels_into_the_report( + import_workflow: ImportModel, +): + recorder = Recorder( + failures={ + Activities.open_import_bundle: ImportGateRejectedError( + 'gate 3 rejected the bundle', step=ImportStep.DECRYPTION, gate=3 + ) + } + ) + patcher, _ = run_with(recorder) + try: + with raises(ImportGateRejectedError): + await import_workflow.run(valid_input()) + finally: + patcher.stop() + + reported = recorder.payload_of(Activities.report_import_pipeline_failure) + assert reported['step'] == ImportStep.DECRYPTION.value + assert reported['gate'] == 3 + + +@mark.asyncio +async def test_a_claim_that_matches_nothing_is_not_reported_twice(import_workflow: ImportModel): + """The claim reports itself, and it runs outside the try — so the channel is hit once.""" + recorder = Recorder( + failures={ + Activities.claim_import_status: ImportLogRowNotClaimableError( + 'row 29 could not be claimed', import_run_id=29 + ) + } + ) + patcher, _ = run_with(recorder) + try: + with raises(ImportLogRowNotClaimableError): + await import_workflow.run(valid_input()) + finally: + patcher.stop() + + assert Activities.report_import_pipeline_failure not in recorder.order + + +@mark.asyncio +async def test_a_report_that_cannot_be_made_does_not_mask_the_failure( + import_workflow: ImportModel, +): + recorder = Recorder( + failures={ + Activities.write_import_model_document: activity_failure('mongo down', 'PyMongoError'), + Activities.report_import_pipeline_failure: RuntimeError('no channel'), + } + ) + patcher, _ = run_with(recorder) + try: + with raises(RuntimeError, match='Activity task failed'): + await import_workflow.run(valid_input()) + finally: + patcher.stop() + + terminal = recorder.payload_of(Activities.record_import_terminal_status) + assert terminal['succeeded'] is False + assert terminal['step'] == ImportStep.MODEL_DOCUMENT.value + + +@mark.asyncio +async def test_a_successful_import_reports_no_failure(import_workflow: ImportModel, workflow_env): + await import_workflow.run(valid_input()) + + assert Activities.report_import_pipeline_failure not in workflow_env.recorder.order + + +@mark.asyncio +async def test_the_report_gives_up_inside_the_short_bound(import_workflow: ImportModel): + """A failure already happened; spending 151 seconds telling the log about it would be worse.""" + recorder = Recorder( + failures={ + Activities.create_import_experiment: activity_failure('refused', 'MlflowException') + } + ) + patcher, _ = run_with(recorder) + try: + with raises(RuntimeError): + await import_workflow.run(valid_input()) + finally: + patcher.stop() + + call = recorder.calls_to(Activities.report_import_pipeline_failure)[0] + assert call.kwargs['retry_policy'] is IMPORT_STATUS_HINT_RETRY_POLICY + assert call.kwargs['start_to_close_timeout'] == STATUS_WRITE_TIMEOUT + + # ------------------------------------------------------------------- the retry split