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

SIENTIAPDE-2013: cap load_latest_data's Mongo fetch at a configurable batch size

Browse Source 
A cold start or notification storm had no backstop: load_latest_data
fetched the entire matching backlog in one query. Confirmed live during
SIENTIAPDE-2012's cluster verification -- an empty Redis cursor plus a
real recurring error let a single run's query exceed Temporal's 4MB
message limit and get stuck retrying forever.

Threads a new notification_queue_batch_size config (env var
NOTIFICATION_QUEUE_BATCH_SIZE, default 5000) through to the repository's
find(), hard cap per run, no drain loop -- the incremental
last_data_timestamp cursor picks up any remainder next run, per the
parent ticket's design decision. Sorts by timestamp ascending so a
capped run always returns the oldest pending documents first; without
that, a capped batch has no guarantee of hitting the oldest documents,
and the cursor would advance past whatever it skipped.

Requires sientia_do>=1.13.0 (adds limit/sort params to
MongoDBRepository.find in both the sync and async client) -- bumped the
pin here per the parent ticket's decision to do it in the same PR.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Eduardo Rios
2026-09-17 09:33:46 -03:00
parent d250383f78
commit 9c153041ad
8 changed files with 79 additions and 6 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 5000), 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
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,

22
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,16 @@ class MongoDB(SientiaMonitoring):
filtering by timestamp to enable incremental data processing. It
handles connection management and provides comprehensive error reporting.
SIENTIAPDE-2013: the query is capped at ``notification_queue_batch_size``
documents (hard cap per run, no drain loop) and sorted by ``timestamp``
ascending, so a capped run always returns the *oldest* pending documents
first. That matters because the caller advances its incremental cursor to
the max timestamp of whatever this returns -- without the ascending sort,
a capped batch could skip over older documents that never get picked up
on a later run. This backstops a cold start, backlog buildup, or
notification storm from fetching an unbounded (and potentially
multi-megabyte) payload in one query.
Args:
input_data (dict[str, Any]): Activity input parameters.
Required fields:
@@ -488,7 +502,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', '5000')),
}

2
requirements.txt
View File

@@ -4,5 +4,5 @@ sqlalchemy
redis
pymongo
jinja2
sientia_do>=1.12.2
sientia_do>=1.13.0
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': 5000,
}