d4a74ec5eb
Update environment configuration and refactor activities to include metrics controller. Remove Kafka settings and adjust Redis and MongoDB initialization. Update tests to reflect changes in initialization and metrics tracking.
117 lines
4.6 KiB
Python
117 lines
4.6 KiB
Python
from temporalio import workflow
|
|
|
|
with workflow.unsafe.imports_passed_through():
|
|
from datetime import timedelta
|
|
from typing import Any
|
|
|
|
from sientia_do.temporal.policies import retry_policy
|
|
|
|
from scouter.activities.activities import Activities
|
|
|
|
|
|
@workflow.defn(name='scouter')
|
|
class Scouter:
|
|
"""
|
|
Main Scouter workflow that orchestrates data ingestion and processing.
|
|
|
|
This workflow serves as the entry point for data processing pipelines. It loads
|
|
data from MongoDB collections, manages data timestamps for incremental processing,
|
|
and delegates the actual data processing to the CoreScouter workflow.
|
|
|
|
The workflow implements a robust data ingestion pattern with:
|
|
- Incremental data loading based on last processed timestamp
|
|
- Automatic timestamp management for data continuity
|
|
- Error handling and retry policies
|
|
- Child workflow orchestration for data processing
|
|
"""
|
|
|
|
@workflow.run
|
|
async def run(self, input_data: dict[str, Any]) -> None:
|
|
"""
|
|
Execute the main Scouter workflow.
|
|
|
|
This method orchestrates the complete data ingestion process:
|
|
1. Retrieves the last processed timestamp from Redis
|
|
2. Loads new data from MongoDB since the last timestamp
|
|
3. Updates the last processed timestamp
|
|
4. Delegates data processing to the CoreScouter workflow
|
|
|
|
Args:
|
|
input_data (dict[str, Any]): Configuration and parameters for the workflow execution.
|
|
Required fields:
|
|
- topic (str): The Kafka topic name for data source identification
|
|
- schedule_name (str): Unique identifier for the data collection schedule
|
|
- model_name (str): Name of the data model being processed
|
|
- model_id (str): Unique identifier for the data model
|
|
- trigger_laborious (bool): Flag to enable intensive data processing
|
|
- filters (dict[str, str]): Data quality filters configuration
|
|
- schema (str): Target database schema for data export
|
|
- table_name (str): Target table name for data export
|
|
- retention_time (int): Data retention period in Redis (seconds)
|
|
- model_tags (dict[str, Any]): Tag-specific configuration including:
|
|
- data_range: [min, max] values for data validation
|
|
- aggr_function: Aggregation method (avg, mdn, max, min, lts)
|
|
- frequency: Data collection frequency in milliseconds
|
|
- topics: List of Kafka topics for data routing
|
|
|
|
Returns:
|
|
None: This workflow doesn't return data, it orchestrates data processing
|
|
|
|
Raises:
|
|
WorkflowExecutionError: If workflow execution fails
|
|
ActivityExecutionError: If any activity fails after retry attempts
|
|
"""
|
|
|
|
input_data['workflow_name'] = 'scouter'
|
|
|
|
metadata = {
|
|
'metadata': {
|
|
'model_id': input_data['model_id'],
|
|
'model_name': input_data['model_name'],
|
|
'schedule_name': input_data['schedule_name'],
|
|
'workflow_name': input_data['workflow_name'],
|
|
}
|
|
}
|
|
|
|
last_data_timestamp = await workflow.execute_local_activity_method(
|
|
Activities.get_last_data_timestamp,
|
|
{
|
|
**metadata,
|
|
'workflow_name': input_data['workflow_name'],
|
|
'schedule_name': input_data['schedule_name'],
|
|
},
|
|
start_to_close_timeout=timedelta(seconds=60),
|
|
retry_policy=retry_policy,
|
|
)
|
|
|
|
data = await workflow.execute_local_activity_method(
|
|
Activities.load_latest_data,
|
|
{
|
|
**metadata,
|
|
'collection_name': f'raw_{input_data["schedule_name"]}',
|
|
'last_data_timestamp': last_data_timestamp,
|
|
},
|
|
start_to_close_timeout=timedelta(seconds=60),
|
|
retry_policy=retry_policy,
|
|
)
|
|
|
|
if not data:
|
|
return
|
|
|
|
await workflow.execute_activity_method(
|
|
Activities.put_last_data_timestamp,
|
|
{
|
|
**metadata,
|
|
'data': data,
|
|
'workflow_name': input_data['workflow_name'],
|
|
'schedule_name': input_data['schedule_name'],
|
|
},
|
|
start_to_close_timeout=timedelta(seconds=60),
|
|
retry_policy=retry_policy,
|
|
)
|
|
|
|
input_data['data'] = data
|
|
input_data['metadata'] = metadata
|
|
|
|
await workflow.execute_child_workflow('subworkflow.core_scouter', input_data)
|