gitea_admin/sientia-dataops-laborious-legacy
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

QTZPOC-20

Browse Source 
QTZPOC-20: Improve structured logging and reporting for all model import pipeline failures by introducing a dedicated activity and centralizing error classification. This ensures consistent observability, with all workflow failures now reported before cleanup. Also, adjusted import step timeouts and refined manual run script cleanup logic.
This commit is contained in:
Vitor Pimentel dos Santos
2026-08-27 14:44:57 -03:00
parent e95e65b72b
commit 853c83e662
18 changed files with 857 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
docs/model-import.md
View File

@@ -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.

1
e2e/import_harness.py
View File

@@ -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',

2
e2e/test_import_model_digest.py
View File

@@ -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]

2
e2e/test_import_model_gates.py
View File

@@ -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]

2
e2e/test_import_model_provisioning_failures.py
View File

@@ -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]

2
e2e/test_import_model_status_record.py
View File

@@ -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

60
laborious/activities/model_import.py
View File

@@ -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:

162
laborious/activities/model_import_errors.py → laborious/utils/models/import_errors.py
View File

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

1
laborious/worker/worker.py
View File

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

59
laborious/workflows/import_model.py
View File

@@ -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],

2
requirements-dev.txt
View File

@@ -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

150
scripts/import_model_manual_run.py
View File

@@ -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')

2
tests/laborious/activities/test_model_import_bundle.py
View File

@@ -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

76
tests/laborious/activities/test_model_import_status.py
View File

@@ -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()

2
tests/laborious/test_import_absence.py
View File

@@ -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'

185
tests/laborious/activities/test_model_import_errors.py → tests/laborious/utils/models/test_import_errors.py
View File

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

2
tests/laborious/worker/test_runtime_task_queues.py
View File

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

185
tests/laborious/workflows/test_import_model.py
View File

@@ -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