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

Merge pull request #43 from Aignosi/fix/SIENTIAPDE-2013

Browse Source 
Fix/sientiapde 2013
This commit is contained in:
Vitor Pimentel dos Santos
2026-09-23 08:38:46 -03:00
committed by GitHub
parent 5cc8015fc7 d3104d66ab
commit d0f204d525
12 changed files with 188 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
CLAUDE.md
View File

@@ -45,7 +45,7 @@ CI quality gate (`.github/workflows/quality-gate.yml`) calls the shared reusable
### Layering
- **`orchestrator/worker/worker.py`** — entrypoint (`python -m orchestrator.worker.worker`). Builds a single `Activities` instance, connects to Temporal, starts Prometheus (`HTTP_METRICS_PORT`) and Temporal SDK metrics (`HTTP_SDK_METRICS_PORT`) servers, then runs three `Worker`s concurrently via `asyncio.gather`, each on its own task queue: `orchestrator-queue`, `alerts-queue`, `reports-queue`. `prepare_worker.py` derives the queue name from the workflow class name (`camel_to_snake`) and reads per-queue concurrency/poller tuning from env vars prefixed with the workflow name (e.g. `ORCHESTRATOR_MAX_CONCURRENT_ACTIVITIES`).
- **`orchestrator/workflows/`** — Temporal workflow definitions: `orchestrator.py` (pipeline/slot/schedule orchestration), `alerts.py` (real-time ERROR notifications), `reports.py` (scheduled all-level reports). `alerts.py` and `reports.py` both delegate to the same two subworkflows in `workflows/subworkflows/`: `load_notification_package.py` (load notifications + receiver configs from Mongo, incremental via Redis timestamp) and `process_notifications.py` (build HTML email, send, log to Postgres). The difference between alerts and reports is which filter activity they call (`filter_notification_alerts` vs `filter_notification_reports`) and their base data filter (`level: ERROR` by default vs all levels) — `load_notification_package.py`'s `_compute_query_time_filter` can widen either beyond that default to the union of active receiver groups' own `levels` (SIENTIAPDE-2011); the ERROR-only default only holds when no active group narrows further.
- **`orchestrator/workflows/`** — Temporal workflow definitions: `orchestrator.py` (pipeline/slot/schedule orchestration), `alerts.py` (real-time ERROR notifications), `reports.py` (scheduled all-level reports). `alerts.py` and `reports.py` both delegate to the same two subworkflows in `workflows/subworkflows/`: `load_notification_package.py` (load notifications + receiver configs from Mongo, incremental via Redis timestamp) and `process_notifications.py` (build HTML email, send, log to Postgres). The difference between alerts and reports is which filter activity they call (`filter_notification_alerts` vs `filter_notification_reports`) and their base data filter (`level: ERROR` by default vs all levels) — `load_notification_package.py`'s `_compute_query_time_filter` can widen either beyond that default to the union of active receiver groups' own `levels` (SIENTIAPDE-2011); the ERROR-only default only holds when no active group narrows further. `load_latest_data`'s Mongo query is capped at `NOTIFICATION_QUEUE_BATCH_SIZE` documents per run (default `0`, meaning unlimited — unchanged from pre-SIENTIAPDE-2013 behavior unless a deployment opts in), sorted by `timestamp` ascending so a capped run always returns the oldest pending batch first (SIENTIAPDE-2013) — a hard cap with no drain loop, since the incremental Redis cursor picks up any remainder on the next run.
- **`orchestrator/activities/activities.py`** — `Activities` is a single class combining every activity mixin via multiple inheritance: `TemporalManager, SlotManager, Formatters, MongoDB, Email, Postgres` (Postgres comes from `sientia_do.temporal.activities.postgres`, the external `sientia-dataops-library` dependency). One instance is constructed in `worker.py` and its bound methods are handed to each `Worker` as the activity list — activities are plain methods on this shared object, not separate classes. Each mixin's `__init__` takes its own config dict plus shared `logger`, `notification_handler`, `metrics_controller`. `Activities.shutdown()` closes every mixin's connections.
- **`orchestrator/utils/`** — `connectors_config.py` builds each config dict from env vars (Temporal/Redis/MongoDB/Postgres/Email); `email_builder.py` renders Jinja2 templates (`utils/templates/email_template.html`, `general_template.html`); `orchestrator_functions.py` transforms pipeline documents into pipeline-type-specific configs (`scouter`, `pi_web_api_scouter`, `predictions_batch`, `minimal_retrain`, `drift`, `simple_metrics`); `converters.py` handles Temporal schedule frequency parsing.

1
e2e/conftest.py
View File

@@ -330,6 +330,7 @@ async def test_activities(
'connection_string': f'mongodb://localhost:{mongo_port}',
'database_name': E2E_DATABASE,
'ttl_index_seconds': 3600,
'notification_queue_batch_size': 0,
},
email_config={
'sender_email': 'e2e@example.com',

2
e2e/scenarios.md
View File

@@ -100,6 +100,8 @@ No notifications → no SMTP, no Postgres row.
- Prior timestamp → only newer notifications returned.
- Query-time filter (SIENTIAPDE-2011) → union of active groups' `levels`, intersection of their `ignore_models_id`; a model ignored by only one of several active groups is not excluded from the fetch.
- Empty Mongo → no Redis timestamp write.
- Batch size cap (SIENTIAPDE-2013) → with 5 pending notifications and a batch size of 2, each run returns exactly the next-oldest 2 (sorted ascending by timestamp), and the Redis cursor advances only to what was actually returned, so three successive runs drain the backlog in order instead of skipping anything.
- Batch size 0 means unlimited (SIENTIAPDE-2013) → the configured default (`0`) does not cap the query; 12 pending notifications all come back in one run, verified against a real MongoDB instance rather than assumed from pymongo's documented convention.
### ProcessNotifications — `e2e/test_subworkflow_process_notifications.py`

111
e2e/test_subworkflow_load_notification_package.py
View File

@@ -228,3 +228,114 @@ async def test_load_package_empty_mongo_no_redis_write(
)
assert redis_client.get('notification_last_timestamp:Alerts') is None
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_load_package_batch_size_caps_and_returns_oldest_first(
temporal_env: WorkflowEnvironment,
orchestrator_worker,
stub_workers,
mongo_uri: str,
redis_client: Redis,
test_activities,
):
"""
SIENTIAPDE-2013: a capped run returns the oldest pending documents first,
and the incremental cursor picks up the remainder on the next run.
Five notifications exist, but the batch size is capped at 2. Each run must
return exactly the next-oldest 2 (never an arbitrary subset), and the
Redis cursor must advance only to the max timestamp of what was actually
returned -- not the true max across all 5 -- or the skipped documents
would never be fetched again.
"""
test_activities.notification_queue_batch_size = 2
seed_receiver_groups(mongo_uri, E2E_DATABASE, [default_receiver_group()])
seed_notifications(
mongo_uri,
E2E_DATABASE,
[
default_notification(
notification_id=f'n{i}',
timestamp=datetime(2024, 6, 1, 12, i, 0, tzinfo=UTC),
)
for i in range(1, 6)
],
)
input_data = load_scenario_input('subworkflow_load_notification_package.json')
_build_metadata(input_data)
first_run = await start_and_await_workflow(
temporal_env.client,
LoadNotificationPackage.run,
input_data,
make_workflow_id('load-pkg-batch-1'),
)
assert [n['notification_id'] for n in first_run['notification_package']] == ['n1', 'n2']
second_run = await start_and_await_workflow(
temporal_env.client,
LoadNotificationPackage.run,
input_data,
make_workflow_id('load-pkg-batch-2'),
)
assert [n['notification_id'] for n in second_run['notification_package']] == ['n3', 'n4']
third_run = await start_and_await_workflow(
temporal_env.client,
LoadNotificationPackage.run,
input_data,
make_workflow_id('load-pkg-batch-3'),
)
assert [n['notification_id'] for n in third_run['notification_package']] == ['n5']
@pytest.mark.e2e
@pytest.mark.asyncio
async def test_load_package_batch_size_zero_means_unlimited(
temporal_env: WorkflowEnvironment,
orchestrator_worker,
stub_workers,
mongo_uri: str,
test_activities,
):
"""
SIENTIAPDE-2013: notification_queue_batch_size=0 (the config default) must
NOT cap the query. This exercises pymongo's own documented convention --
`Collection.find(..., limit=0)` is equivalent to no limit at all -- against
a real MongoDB instance rather than trusting the convention on faith.
12 notifications is deliberately more than any other test in this file
seeds, so a regression that silently caps at some smaller default (e.g. if
`limit=0` were ever misread as "return nothing", or pymongo's own behavior
changed) would fail loudly here instead of passing by coincidence.
"""
test_activities.notification_queue_batch_size = 0
seed_receiver_groups(mongo_uri, E2E_DATABASE, [default_receiver_group()])
seed_notifications(
mongo_uri,
E2E_DATABASE,
[
default_notification(
notification_id=f'n{i}',
timestamp=datetime(2024, 6, 1, 12, i, 0, tzinfo=UTC),
)
for i in range(1, 13)
],
)
input_data = load_scenario_input('subworkflow_load_notification_package.json')
_build_metadata(input_data)
result = await start_and_await_workflow(
temporal_env.client,
LoadNotificationPackage.run,
input_data,
make_workflow_id('load-pkg-batch-unlimited'),
)
assert len(result['notification_package']) == 12

1
orchestrator/activities/activities.py
View File

@@ -84,6 +84,7 @@ class Activities(TemporalManager, SlotManager, Formatters, MongoDB, Email, Postg
connection_string=mongodb_config['connection_string'],
database_name=mongodb_config['database_name'],
ttl_index_seconds=mongodb_config['ttl_index_seconds'],
notification_queue_batch_size=mongodb_config['notification_queue_batch_size'],
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,

16
orchestrator/activities/mongo_db.py
View File

@@ -31,6 +31,8 @@ class MongoDB(SientiaMonitoring):
ttl_index_seconds (int): TTL index duration in seconds
logger (Logger): Application logger instance
notification_handler (NotificationHandler): Notification management handler
notification_queue_batch_size (int): Hard cap on documents fetched per
`load_latest_data` run (SIENTIAPDE-2013). `0` means no limit.
"""
def __init__(
@@ -41,6 +43,7 @@ class MongoDB(SientiaMonitoring):
logger: Logger,
notification_handler: NotificationHandler,
metrics_controller: MetricsController,
notification_queue_batch_size: int = 0,
):
SientiaMonitoring.__init__(
self,
@@ -61,6 +64,7 @@ class MongoDB(SientiaMonitoring):
)
self.ttl_index_seconds = ttl_index_seconds
self.notification_queue_batch_size = notification_queue_batch_size
# Initialize MongoDB client here (omitted for brevity)
logger.info('MongoDB connection initialized')
@@ -447,6 +451,10 @@ class MongoDB(SientiaMonitoring):
filtering by timestamp to enable incremental data processing. It
handles connection management and provides comprehensive error reporting.
The query is capped at ``notification_queue_batch_size`` documents and sorted by
``timestamp`` ascending, so a capped run returns the oldest pending documents first --
the caller advances its cursor only to what was actually returned, so nothing is skipped.
Args:
input_data (dict[str, Any]): Activity input parameters.
Required fields:
@@ -488,7 +496,13 @@ class MongoDB(SientiaMonitoring):
self.debug(f'Data filter: {data_filter}', metadata=metadata)
data = self.mongo_db_repository.find(collection_name, data_filter, metadata)
data = self.mongo_db_repository.find(
collection_name,
data_filter,
metadata,
limit=self.notification_queue_batch_size,
sort=[('timestamp', 1)],
)
self.debug(f'Collected: {data}', metadata=metadata)

4
orchestrator/utils/connectors_config.py
View File

@@ -21,7 +21,8 @@ def build_mongodb_config():
Build MongoDB configuration from environment variables.
Returns:
dict: MongoDB configuration with connection string, database name, and TTL index seconds.
dict: MongoDB configuration with connection string, database name, TTL index seconds,
and the notification queue's per-run batch size cap.
"""
username = getenv('MONGODB_USERNAME', 'root')
password = getenv('MONGODB_PASSWORD', 'wKZDbMNU1c')
@@ -32,6 +33,7 @@ def build_mongodb_config():
'connection_string': connection_string,
'database_name': getenv('MONGODB_DATABASE_NAME', 'sientia'),
'ttl_index_seconds': int(getenv('MONGODB_TTL_INDEX_HOURS', '1')) * 3600,
'notification_queue_batch_size': int(getenv('NOTIFICATION_QUEUE_BATCH_SIZE', '0')),
}

2
requirements-local.txt
View File

@@ -4,5 +4,5 @@ sqlalchemy
redis
pymongo
jinja2
git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.12.2
git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.12.3
prometheus-client

2
requirements.txt
View File

@@ -4,5 +4,5 @@ sqlalchemy
redis
pymongo
jinja2
sientia_do>=1.12.2
sientia_do>=1.12.3
prometheus-client

2
tests/orchestrator/activities/test_activities.py
View File

@@ -27,6 +27,7 @@ def test___init__(
'connection_string': 'mongodb://localhost:27017',
'database_name': 'test_db',
'ttl_index_seconds': 3600,
'notification_queue_batch_size': 5000,
}
redis_config = {'host': 'localhost', 'port': 6379, 'username': 'admin', 'password': 'password'}
@@ -89,6 +90,7 @@ def test___init__(
connection_string='mongodb://localhost:27017',
database_name='test_db',
ttl_index_seconds=3600,
notification_queue_batch_size=5000,
logger=logger,
notification_handler=notification_handler,
metrics_controller=mock_metrics_controller.return_value,

48
tests/orchestrator/activities/test_mongo_db.py
View File

@@ -31,13 +31,14 @@ def test___init__(mongo_mock):
notification_handler = MagicMock()
metrics_controller = AsyncMock()
MongoDB(
mongo = MongoDB(
connection_string='mongodb://localhost:27017',
database_name='test_db',
ttl_index_seconds=3600,
logger=logger,
notification_handler=notification_handler,
metrics_controller=metrics_controller,
notification_queue_batch_size=5000,
)
mongo_mock.assert_called_once_with(
@@ -47,6 +48,22 @@ def test___init__(mongo_mock):
notification_handler=notification_handler,
metrics_controller=metrics_controller,
)
assert mongo.notification_queue_batch_size == 5000
@patch('orchestrator.activities.mongo_db.MongoDBRepository')
def test___init___notification_queue_batch_size_defaults_to_zero(mongo_mock):
"""0 means no limit, matching the underlying find()'s own convention."""
mongo = MongoDB(
connection_string='mongodb://localhost:27017',
database_name='test_db',
ttl_index_seconds=3600,
logger=MagicMock(),
notification_handler=MagicMock(),
metrics_controller=AsyncMock(),
)
assert mongo.notification_queue_batch_size == 0
@patch('orchestrator.activities.mongo_db.SientiaMonitoring')
@@ -480,6 +497,8 @@ def test_load_latest_data_none_last_data_timestamp(mongo_db):
'test_collection',
{'level': 'ERROR'},
{'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'},
limit=0,
sort=[('timestamp', 1)],
)
assert result == [
@@ -487,6 +506,29 @@ def test_load_latest_data_none_last_data_timestamp(mongo_db):
]
def test_load_latest_data_passes_configured_batch_size_as_limit(mongo_db):
"""SIENTIAPDE-2013: the configured notification_queue_batch_size caps the query."""
mongo_db.notification_queue_batch_size = 100
mongo_db.mongo_db_repository.find = MagicMock(return_value=[])
mongo_db.load_latest_data(
{
'metadata': {'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'},
'collection_name': 'test_collection',
'last_data_timestamp': None,
'base_data_filter': {'level': 'ERROR'},
}
)
mongo_db.mongo_db_repository.find.assert_called_once_with(
'test_collection',
{'level': 'ERROR'},
{'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'},
limit=100,
sort=[('timestamp', 1)],
)
def test_load_latest_data_not_none_last_data_timestamp(mongo_db):
"""Test load_latest_data compares native datetime against native datetime in $gt"""
@@ -516,6 +558,8 @@ def test_load_latest_data_not_none_last_data_timestamp(mongo_db):
'timestamp': {'$gt': datetime(2023, 1, 1, 12, 0, 0, tzinfo=UTC)},
},
{'workflow_name': 'test_pipeline', 'schedule_name': 'test_schedule'},
limit=0,
sort=[('timestamp', 1)],
)
assert result == [
@@ -529,7 +573,7 @@ def test_load_latest_data_run_boundary_no_skip_or_duplicate(mongo_db):
boundary_timestamp = datetime(2023, 1, 1, 12, 0, 0, tzinfo=UTC)
def fake_find(collection_name, data_filter, metadata):
def fake_find(collection_name, data_filter, metadata, limit=0, sort=None):
gt = data_filter.get('timestamp', {}).get('$gt')
all_docs = [
{

4
tests/orchestrator/utils/test_connectors_config.py
View File

@@ -64,10 +64,12 @@ def test_build_mongo_db_config_with_env_vars():
environ['MONGODB_URL'] = 'localhost:27018'
environ['MONGODB_DATABASE_NAME'] = 'test_db'
environ['MONGODB_TTL_INDEX_HOURS'] = '2'
environ['NOTIFICATION_QUEUE_BATCH_SIZE'] = '2000'
assert build_mongodb_config() == {
'connection_string': 'mongodb://sientia1:sientia1@localhost:27018',
'database_name': 'test_db',
'ttl_index_seconds': 7200,
'notification_queue_batch_size': 2000,
}
@@ -77,10 +79,12 @@ def test_build_mongo_db_config_with_defaults():
environ.pop('MONGODB_DATABASE_NAME', None)
environ.pop('MONGODB_URL', None)
environ.pop('MONGODB_TTL_INDEX_HOURS', None)
environ.pop('NOTIFICATION_QUEUE_BATCH_SIZE', None)
assert build_mongodb_config() == {
'connection_string': 'mongodb://root:wKZDbMNU1c@localhost:27018',
'database_name': 'sientia',
'ttl_index_seconds': 3600,
'notification_queue_batch_size': 0,
}