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

feature-rc/SIENTIAPDE-2164 - refactor: centralize prediction metadata write in a dedicated activity; REPEAT now writes metadata too

Browse Source
This commit is contained in:
Carla Pinto
2026-10-01 17:46:20 +02:00
parent db32450c68
commit 9c628e70a6
11 changed files with 606 additions and 217 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
laborious/activities/activities.py
View File

@@ -17,18 +17,21 @@ with workflow.unsafe.imports_passed_through():
from laborious.activities.mlflow import MLFlow
from laborious.activities.model_metrics import ModelMetrics
from laborious.activities.opc import OPC
from laborious.activities.prediction_metadata import PredictionMetadata
from laborious.activities.storage import Storage
from laborious.utils.connectors_config import build_mlflow_config
class Activities(Storage, MLFlow, Gates, OPC, ModelMetrics, API):
class Activities(Storage, MLFlow, Gates, OPC, ModelMetrics, API, PredictionMetadata):
"""
Central orchestrator for all Temporal activities used by Laborious workflows.
Composes Storage (Postgres + MinIO offload), MLFlow (wrapper-based inference and retrain
via ``SientiaMLflowRepository``), Gates (data quality and ML response filters), OPC exports,
drift/simple metrics, and PI Web API writes. The worker constructs one ``Activities`` instance
per process and registers its callables on multiple workers bound to different task queues.
drift/simple metrics, PI Web API writes, and PredictionMetadata (the single write point for
gate metadata — see `laborious.activities.prediction_metadata`). The worker constructs one
``Activities`` instance per process and registers its callables on multiple workers bound to
different task queues.
MLflow connectivity: unless ``mlflow_repository`` is injected (tests only), this class builds
``SientiaMLflowRepository`` from ``build_mlflow_config()`` so tracking credentials and URL
@@ -70,7 +73,7 @@ class Activities(Storage, MLFlow, Gates, OPC, ModelMetrics, API):
- metrics_controller: Optional shared controller; if ``None``, a new one is created.
- mlflow_repository: Optional ``SientiaMLflowRepository`` for unit/e2e tests; in production
leave unset so the repository is built from environment via ``build_mlflow_config()``.
- mongo_repository: Optional ``MongoDBRepository`` (gate metadata, best-effort) for
- mongo_repository: Optional ``MongoDBRepository`` (prediction metadata, best-effort) for
unit/e2e tests; in production leave unset so the repository is built from environment
via ``build_mongodb_config()``, same pattern as ``mlflow_repository`` above.
@@ -145,6 +148,13 @@ class Activities(Storage, MLFlow, Gates, OPC, ModelMetrics, API):
Gates.__init__(
self,
minio_repository=minio_repository,
logger=logger,
notification_handler=notification_handler,
metrics_controller=mc,
)
PredictionMetadata.__init__(
self,
mongo_repository=mongo_repository,
logger=logger,
notification_handler=notification_handler,
@@ -192,3 +202,4 @@ class Activities(Storage, MLFlow, Gates, OPC, ModelMetrics, API):
OPC.close(self)
ModelMetrics.close(self)
API.close(self)
PredictionMetadata.close(self)

101
laborious/activities/gates.py
View File

@@ -5,7 +5,6 @@ with workflow.unsafe.imports_passed_through():
from collections.abc import Callable, Mapping
from typing import Any
from bson import ObjectId
from pandas import DataFrame
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
from sientia_do.notifications.models import NotificationLevel
@@ -13,7 +12,6 @@ with workflow.unsafe.imports_passed_through():
from sientia_do.observability.metrics_controller import MetricsController
from sientia_do.observability.sientia_monitoring import SientiaMonitoring
from sientia_do.repository.minio_repository_sync import MinioRepository
from sientia_do.repository.mongodb_repository_sync import MongoDBRepository
from sientia_do.utils.formatters import create_sample_dict
from laborious import metrics
@@ -125,14 +123,11 @@ class Gates(SientiaMonitoring):
"""
minio_repository: MinioRepository | None = None
mongo_repository: MongoDBRepository | None = None
_MAX_DEBUG_DATAFRAME_ROWS = 100
_PREDICTION_METADATA_COLLECTION = 'prediction_metadata'
def __init__(
self,
minio_repository: MinioRepository | None = None,
mongo_repository: MongoDBRepository | None = None,
logger: Logger | None = None,
notification_handler: NotificationHandler | None = None,
metrics_controller: MetricsController | None = None,
@@ -143,14 +138,11 @@ class Gates(SientiaMonitoring):
Args:
logger: Logger instance for observability and debugging
notification_handler: Notification handler for alerts and monitoring
mongo_repository: MongoDB, where gate metadata (triggered filter, policy, counter) is
written best-effort. `None` is a valid state (see `_write_prediction_metadata`).
Raises:
Exception: If BaseActivity initialization fails
"""
self.minio_repository = minio_repository
self.mongo_repository = mongo_repository
SientiaMonitoring.__init__(
self,
logger=logger,
@@ -168,11 +160,6 @@ class Gates(SientiaMonitoring):
self.minio_repository.close()
finally:
self.minio_repository = None
if self.mongo_repository is not None:
try:
self.mongo_repository.close()
finally:
self.mongo_repository = None
SientiaMonitoring.shutdown(self)
def __del__(self):
@@ -196,64 +183,6 @@ class Gates(SientiaMonitoring):
metadata,
)
def _write_prediction_metadata(
self, document: dict[str, Any] | None, metadata: dict[str, Any]
) -> str | None:
"""
Best-effort write of gate metadata (triggered filter, policy, counter) to Mongo.
Implements both business rules for this feature in one place:
- The prediction write must never be blocked or failed by a metadata write. A failed
`insert(...)` is swallowed, never re-raised, logged here with gate-level context (which
`mongo_repository.insert(...)` itself can't know) — matching how `storage.py`'s
`cleanup_minio_objects_expired` also logs again at the call site even though the
repository already logs internally and emits its own Prometheus metric
(`MONGODB_WRITE_ERROR_COUNT`); the generic library log is for the metric/trace, this
one is for whoever is reading logs for this specific gate.
- Nothing is written when there's nothing relevant to report. A falsy `document` (e.g.
no filter triggered) skips Mongo entirely, so callers can call this unconditionally
and get `None` back for "no metadata this time" instead of deciding upfront whether
there's something worth saving.
`insert(...)` doesn't hand back the inserted document's `_id`, so it is generated here,
client-side, before the insert, as a plain string (not a BSON ObjectId) — matching this
codebase's convention of string `_id`s (e.g. the `models` collection) and the 24-hex-char
size of the `predictions.metadata_id` Postgres column that will reference it.
Caller contract (not enforced here — no caller exists yet in this codebase): only call
this when the gate's resolved `path_flag` is `CONTINUE`. `STOP` terminates the workflow
before any `predictions` row is written (`prediction_process.py`'s `path_flag_handler`);
`REPEAT` reuses the last existing row instead of writing a new one. In both cases, a
document written here would be orphaned — no `predictions.metadata_id` will ever
reference it. Only `CONTINUE` reaches `format_and_export_prediction`, the one path that
writes a new `predictions` row.
Args:
document: Metadata payload (filter, policy, counter, etc.); `None`/empty skips the
write entirely. No `_id` key expected.
metadata: Workflow metadata for contextual logging.
Return:
The generated id, or `None` when there's nothing to write, Mongo isn't configured, or
the insert failed.
"""
if not document or self.mongo_repository is None:
return None
generated_id = str(ObjectId())
try:
self.mongo_repository.insert(
self._PREDICTION_METADATA_COLLECTION,
{**document, '_id': generated_id},
metadata,
)
except Exception as e:
self.error(f'Failed to write prediction metadata: {e}', metadata)
return None
return generated_id
@staticmethod
def _read_filter_entry(config: dict[str, Any]) -> tuple[str, dict[str, Any]]:
"""
@@ -271,7 +200,9 @@ class Gates(SientiaMonitoring):
return policy, filter_config
@activity.defn(name='input_gate')
def input_gate(self, input_data: dict[str, Any]) -> tuple[str | None, int, str, str | None]:
def input_gate(
self, input_data: dict[str, Any]
) -> tuple[str | None, int, str, dict[str, Any] | None]:
"""
Apply input data quality filters and validation.
@@ -286,8 +217,9 @@ class Gates(SientiaMonitoring):
3. Determines appropriate path decisions (STOP, CONTINUE, REPEAT)
4. Provides confidence scores and detailed comments
5. Handles errors gracefully with notification integration
6. When the resolved path is CONTINUE, best-effort writes which filter(s) triggered,
the policy and a counter to `prediction_metadata` (see `_write_prediction_metadata`)
6. Builds a metadata document describing which filter(s) triggered, with their policy
and a counter, whenever one does — this activity only computes it, it never writes it
(see `PredictionMetadata.write_prediction_metadata` for the actual write)
Args:
input_data: Configuration and data for input validation
@@ -298,14 +230,14 @@ class Gates(SientiaMonitoring):
- path_priority (list[str]): Priority order for path decisions
Returns:
tuple: (path_flag, confidence, comment, metadata_id)
tuple: (path_flag, confidence, comment, document)
- path_flag (str | None): Decision path (STOP, CONTINUE, REPEAT, or None)
- confidence (int): Confidence score for the decision
- comment (str): Detailed explanation of the decision
- metadata_id (str | None): Id of the `prediction_metadata` document written for
this gate's decision; `None` unless `path_flag` is `CONTINUE` (STOP/REPEAT
never write a new `predictions` row, so writing metadata for them would
orphan it — see `_write_prediction_metadata`'s caller contract)
- document (dict | None): Metadata describing the filter(s) that produced
`path_flag`, shaped `{'filters': [{'name', 'policy', 'counter'}, ...]}`;
`None` when nothing was filtered. Not written anywhere by this activity —
the caller decides whether and how to persist it.
Raises:
Exception: If filter execution fails or configuration is invalid
@@ -373,17 +305,12 @@ class Gates(SientiaMonitoring):
for fil, filter_config in matched
],
}
metadata_id = (
self._write_prediction_metadata(document, metadata)
if path_flag == 'CONTINUE'
else None
)
return (
path_flag,
input_path_confidence[path_flag],
'Input data with bad quality',
metadata_id,
document,
)
self.info('Nothing was filtered by the input gate', metadata)
@@ -786,6 +713,9 @@ class Gates(SientiaMonitoring):
- model_id (str): Unique identifier for the ML model
- prediction_confidence (float): Confidence score (typically low for errors)
- comment (str): Error description or operational comment
- metadata_id (str | None, optional): Id of the `prediction_metadata` Mongo
document to reference for this row; `None`/absent when there's nothing to
reference
Returns:
dict: Formatted default prediction data with error indicators
@@ -803,6 +733,7 @@ class Gates(SientiaMonitoring):
'prediction_confidence': [input_data['prediction_confidence']],
'prediction_status': ['Bad'],
'comments': [input_data['comment']],
'metadata_id': [input_data.get('metadata_id')],
}
)

132
laborious/activities/prediction_metadata.py Normal file
View File

@@ -0,0 +1,132 @@
from temporalio import activity, workflow
with workflow.unsafe.imports_passed_through():
from typing import Any
from bson import ObjectId
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
from sientia_do.observability.logger import Logger
from sientia_do.observability.metrics_controller import MetricsController
from sientia_do.observability.sientia_monitoring import SientiaMonitoring
from sientia_do.repository.mongodb_repository_sync import MongoDBRepository
class PredictionMetadata(SientiaMonitoring):
"""
Centralized, best-effort writer for prediction gate metadata (triggered filter, policy,
counter) to MongoDB.
Gates only compute and return a metadata document; they never write it themselves. Keeping
the write here, instead of in each gate, puts the one real write point in the subworkflow
that's about to persist a `predictions` row — the only place that reliably knows whether one
is being written at all — and lets other metadata sources reuse the same activity instead of
duplicating write logic wherever they originate.
"""
mongo_repository: MongoDBRepository | None = None
_PREDICTION_METADATA_COLLECTION = 'prediction_metadata'
def __init__(
self,
mongo_repository: MongoDBRepository | None = None,
logger: Logger | None = None,
notification_handler: NotificationHandler | None = None,
metrics_controller: MetricsController | None = None,
):
"""
Initialize the prediction metadata writer.
Args:
mongo_repository: MongoDB, where gate metadata (triggered filter, policy, counter) is
written best-effort. `None` is a valid state (see `write_prediction_metadata`).
Raises:
Exception: If BaseActivity initialization fails
"""
self.mongo_repository = mongo_repository
SientiaMonitoring.__init__(
self,
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
)
def close(self) -> None:
"""
Close the prediction metadata writer and clean up resources.
"""
if self.mongo_repository is not None:
try:
self.mongo_repository.close()
finally:
self.mongo_repository = None
SientiaMonitoring.shutdown(self)
def __del__(self):
self.close()
@activity.defn(name='write_prediction_metadata')
def write_prediction_metadata(self, input_data: dict[str, Any]) -> str | None:
"""
Best-effort write of one or more gate metadata documents to Mongo, combined into a single
`prediction_metadata` document.
Implements both business rules for this feature in one place:
- The prediction write must never be blocked or failed by a metadata write. A failed
`insert(...)` is swallowed, never re-raised, logged here with call-site context (which
`mongo_repository.insert(...)` itself can't know) — matching how `storage.py`'s
`cleanup_minio_objects_expired` also logs again at the call site even though the
repository already logs internally and emits its own Prometheus metric
(`MONGODB_WRITE_ERROR_COUNT`); the generic library log is for the metric/trace, this
one is for whoever is reading logs for this specific call.
- Nothing is written when there's nothing relevant to report. An empty/missing
`documents` list (or one full of falsy/empty entries) skips Mongo entirely, so callers
can call this unconditionally and get `None` back for "no metadata this time" instead
of deciding upfront whether there's something worth saving.
Takes a *list* of documents, not a single one: each caller contributes its own document
(shaped like `{'filters': [...]}`), and this flattens every `filters` entry across all of
them into one `prediction_metadata` document before the single insert. Today only one
gate ever contributes a document per prediction cycle (the gate chain short-circuits on
the first non-`None` `path_flag`), but the list shape means this activity won't need to
change again once more than one source contributes metadata for the same cycle.
`insert(...)` doesn't hand back the inserted document's `_id`, so it is generated here,
client-side, before the insert, as a plain string (not a BSON ObjectId) — matching this
codebase's convention of string `_id`s (e.g. the `models` collection) and the 24-hex-char
size of the `predictions.metadata_id` Postgres column that will reference it.
Args:
input_data: Dict with:
- documents (list[dict[str, Any]]): Metadata payloads to combine, each shaped
like `{'filters': [...]}`. Missing/empty, or entries with no `filters`, skip
the write entirely.
- metadata (dict): Workflow metadata for contextual logging.
Return:
The generated id, or `None` when there's nothing to write, Mongo isn't configured, or
the insert failed.
"""
metadata = input_data['metadata']
documents = input_data.get('documents') or []
filters = [
entry for document in documents if document for entry in document.get('filters', [])
]
if not filters or self.mongo_repository is None:
return None
generated_id = str(ObjectId())
try:
self.mongo_repository.insert(
self._PREDICTION_METADATA_COLLECTION,
{'filters': filters, '_id': generated_id},
metadata,
)
except Exception as e:
self.error(f'Failed to write prediction metadata: {e}', metadata)
return None
return generated_id

2
laborious/worker/worker.py
View File

@@ -249,6 +249,8 @@ async def main():
activities.format_transformed_data,
activities.format_prediction,
activities.format_default_prediction,
# Prediction metadata
activities.write_prediction_metadata,
# OPC
activities.write_opc_data,
# Postgres / MinIO offload

6
laborious/workflows/sub_workflows/format_and_export_prediction.py
View File

@@ -75,6 +75,11 @@ class FormatAndExportPrediction:
Required when path_flag is None
- comment (str): Operational comment or error description
Required when path_flag is not None
- metadata_id (str | None): Id of the `prediction_metadata` Mongo document to
reference for this row, already written by the caller; `None` when there's
nothing to reference. Only meaningful when path_flag is not None — the
normal (path_flag is None) path never has metadata to reference, so this
key is irrelevant there.
Returns:
None: The workflow completes successfully when all export operations finish
@@ -154,6 +159,7 @@ class FormatAndExportPrediction:
'model_id': input_data['model_id'],
'prediction_confidence': prediction_confidence,
'comment': input_data['comment'],
'metadata_id': input_data.get('metadata_id'),
},
retry_policy=retry_policy,
start_to_close_timeout=timedelta(seconds=60),

61
laborious/workflows/sub_workflows/prediction_process.py
View File

@@ -132,27 +132,28 @@ class PredictionProcess:
'path_priority': input_data['path_priority'],
}
# `_metadata_id` (prediction_metadata document id, only non-None when path_flag ends up
# CONTINUE) isn't threaded through yet — SIENTIAPDE-2167 wires it to Postgres. Star-unpack
# rather than a fixed 4-tuple: `input_gate` is a local activity, so its result is recorded
# in workflow history; an execution already in flight when this field was added would
# replay a 3-element result here, and exact unpacking would raise ValueError on replay
# (review finding, 2026-09-30).
# `document` describes which filter(s) produced `path_flag` (`None` when nothing was
# filtered); `path_flag_handler` is the one that decides whether and how to persist it.
# Star-unpack rather than a fixed 4-tuple: `input_gate` is a local activity, so its result
# is recorded in workflow history; an execution already in flight when this field was
# added would replay a 3-element result here, and exact unpacking would raise ValueError
# on replay (review finding, 2026-09-30).
(
path_flag,
confidence,
comment,
*_metadata_id,
*document_tail,
) = await workflow.execute_local_activity_method(
Activities.input_gate,
gate_input,
retry_policy=retry_policy,
start_to_close_timeout=timedelta(minutes=1),
)
document = document_tail[0] if document_tail else None
# Handle path decision based on filter results
if await self.path_flag_handler(
data, path_flag, input_data, confidence, last_timestamp, comment
data, path_flag, input_data, confidence, last_timestamp, comment, document
):
return
@@ -258,6 +259,33 @@ class PredictionProcess:
},
)
async def _write_prediction_metadata(
self, metadata: dict[str, Any], document: dict[str, Any] | None
) -> str | None:
"""
Write a gate's metadata document via the centralized write activity.
Shared by the CONTINUE and REPEAT branches of `path_flag_handler` — both persist a new
`predictions` row (`REPEAT` reuses the previous row's values but still inserts a new
one), so both need a `metadata_id` to reference it.
Args:
metadata: Workflow metadata for contextual logging.
document: Metadata produced by the gate that resolved this `path_flag`; `None` when
nothing was filtered, or when the gate that produced this `path_flag` doesn't
build one yet.
Return:
The generated id, or `None` when there's nothing to write (see
`PredictionMetadata.write_prediction_metadata`).
"""
return await workflow.execute_activity_method(
Activities.write_prediction_metadata,
{**metadata, 'documents': [document] if document else []},
retry_policy=retry_policy,
start_to_close_timeout=timedelta(minutes=1),
)
async def path_flag_handler(
self,
data: dict[str, Any],
@@ -266,6 +294,7 @@ class PredictionProcess:
confidence: int,
last_timestamp: str,
comment: str,
document: dict[str, Any] | None = None,
) -> bool:
"""
Handle path decisions based on filter results and confidence levels.
@@ -291,6 +320,13 @@ class PredictionProcess:
confidence: Confidence level from filter validation
last_timestamp: Last processed timestamp
comment: Additional information about the filter result
document: Metadata describing the filter(s) that produced `path_flag`, as returned by
the gate (`None` when nothing was filtered, or when the gate that produced this
`path_flag` doesn't build one yet). Written via `write_prediction_metadata` for
`CONTINUE` and `REPEAT` alike — both persist a `predictions` row (`REPEAT` reuses
the previous row's values but still inserts a new one), so both can reference the
resulting `metadata_id`; `STOP` never reaches a write at all, so its `document` is
simply never used here.
Returns:
bool: True if processing should stop, False to continue
@@ -315,7 +351,11 @@ class PredictionProcess:
# Stop processing and exit workflow
return True
elif path_flag == 'REPEAT':
# Repeat last prediction if available
metadata_id = await self._write_prediction_metadata(metadata, document)
# Repeat last prediction if available. `metadata_id` isn't consumed by
# `repeat_last_prediction` yet (its INSERT...SELECT has no `metadata_id` column —
# pending a change in `sientia-dataops-library`); passed through already so no further
# change is needed here once that lands.
await workflow.execute_activity_method(
Activities.repeat_last_prediction,
{
@@ -324,12 +364,14 @@ class PredictionProcess:
'table_name': table_name,
'model': model_id,
'last_timestamp': last_timestamp,
'metadata_id': metadata_id,
},
retry_policy=retry_policy,
start_to_close_timeout=timedelta(minutes=1),
)
return True
elif path_flag == 'CONTINUE':
metadata_id = await self._write_prediction_metadata(metadata, document)
# call write workflow
await workflow.execute_child_workflow(
'subworkflow.format_and_export_prediction',
@@ -346,6 +388,7 @@ class PredictionProcess:
'table_name': table_name,
'transform_table_name': transform_table_name,
'comment': comment,
'metadata_id': metadata_id,
'opc_output_config': input_data['opc_output_config'],
'pi_web_api_output_config': input_data['pi_web_api_output_config'],
'prediction_store_policy': input_data['prediction_store_policy'],