diff --git a/orchestrator/activities/formatters.py b/orchestrator/activities/formatters.py index 9b7ce92..636d8a9 100644 --- a/orchestrator/activities/formatters.py +++ b/orchestrator/activities/formatters.py @@ -2,10 +2,10 @@ from temporalio import activity, workflow with workflow.unsafe.imports_passed_through(): import json - from collections.abc import Hashable + from collections.abc import Callable, Hashable from logging import Logger from math import ceil - from typing import Any + from typing import Any, TypedDict from pandas import DataFrame from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler @@ -18,6 +18,7 @@ with workflow.unsafe.imports_passed_through(): drift, gather_read_tags, minimal_retrain, + pi_web_api_scouter, predictions_batch, scouter, simple_metrics, @@ -26,6 +27,68 @@ with workflow.unsafe.imports_passed_through(): topic_separator = '\n ========== \n' +class ScheduleType(TypedDict): + """ + Type definition for schedule type configuration entries. + + Each schedule type entry maps a workflow type string to its corresponding + namespace and configuration builder function. + + Attributes: + namespace (str): The Temporal namespace where workflows of this type execute. + Valid values: 'scouter', 'laborious' + function (Callable): Configuration builder function that transforms pipeline + configuration into Temporal-compatible workflow arguments + """ + + namespace: str + function: Callable + + +schedule_types: dict[str, ScheduleType] = { + 'scouter': { + 'namespace': 'scouter', + 'function': scouter, + }, + 'pi_web_api_scouter': { + 'namespace': 'scouter', + 'function': pi_web_api_scouter, + }, + 'predictions_batch': { + 'namespace': 'laborious', + 'function': predictions_batch, + }, + 'minimal_retrain': { + 'namespace': 'laborious', + 'function': minimal_retrain, + }, + 'drift': { + 'namespace': 'laborious', + 'function': drift, + }, + 'simple_metrics': { + 'namespace': 'laborious', + 'function': simple_metrics, + }, +} +""" +Registry mapping workflow types to their namespace and configuration builder functions. + +Supported workflow types: + - scouter: OPC data collection using OPC UA protocol + - pi_web_api_scouter: Data collection using PI Web API + - predictions_batch: ML model prediction workflows with OPC write-back + - minimal_retrain: Model retraining workflows using SQL queries + - drift: Data drift detection and monitoring workflows + - simple_metrics: Model performance metrics computation workflows + +Each entry specifies: + - namespace: Target Temporal namespace for workflow execution + - function: Configuration builder that transforms MongoDB pipeline config + into Temporal workflow arguments +""" + + class Formatters(SientiaMonitoring): """ Schedule and slot configuration formatting and notification filtering activity. @@ -115,41 +178,22 @@ class Formatters(SientiaMonitoring): } for pipeline in pipelines: - if pipeline['workflow_type'] == 'scouter': - schedule_config[self.scouter_namespace][pipeline['schedule_name']] = { - **scouter(pipeline), - 'updated_at': pipeline.get( - 'updated_at', now().strftime(DATETIME_FORMAT_MS_WITH_TZ) - ), - } - elif pipeline['workflow_type'] == 'predictions_batch': - schedule_config[self.laborious_namespace][pipeline['schedule_name']] = { - **predictions_batch(pipeline), - 'updated_at': pipeline.get( - 'updated_at', now().strftime(DATETIME_FORMAT_MS_WITH_TZ) - ), - } - elif pipeline['workflow_type'] == 'minimal_retrain': - schedule_config[self.laborious_namespace][pipeline['schedule_name']] = { - **minimal_retrain(pipeline), - 'updated_at': pipeline.get( - 'updated_at', now().strftime(DATETIME_FORMAT_MS_WITH_TZ) - ), - } - elif pipeline['workflow_type'] == 'drift': - schedule_config[self.laborious_namespace][pipeline['schedule_name']] = { - **drift(pipeline), - 'updated_at': pipeline.get( - 'updated_at', now().strftime(DATETIME_FORMAT_MS_WITH_TZ) - ), - } - elif pipeline['workflow_type'] == 'simple_metrics': - schedule_config[self.laborious_namespace][pipeline['schedule_name']] = { - **simple_metrics(pipeline), - 'updated_at': pipeline.get( - 'updated_at', now().strftime(DATETIME_FORMAT_MS_WITH_TZ) - ), - } + workflow_type = pipeline['workflow_type'] + + if workflow_type not in schedule_types: + self.error(f'Workflow type {workflow_type} not supported', metadata=metadata) + continue + + schedule_type = schedule_types[workflow_type] + namespace = schedule_type['namespace'] + function = schedule_type['function'] + + schedule_config[namespace][pipeline['schedule_name']] = { + **function(pipeline), + 'updated_at': pipeline.get( + 'updated_at', now().strftime(DATETIME_FORMAT_MS_WITH_TZ) + ), + } self.info('Processed schedules', metadata=metadata) self.debug(json.dumps(schedule_config, indent=4, sort_keys=True), metadata=metadata) diff --git a/orchestrator/activities/mongo_db.py b/orchestrator/activities/mongo_db.py index 0339ab5..4ae3a96 100644 --- a/orchestrator/activities/mongo_db.py +++ b/orchestrator/activities/mongo_db.py @@ -393,7 +393,9 @@ class MongoDB(SientiaMonitoring): created_indexes = [] for _pipeline_name, pipeline_config in pipelines.items(): - collection = pipeline_config['topic'] + collection = pipeline_config.get('topic', None) + if not collection: + continue try: # Check if collection exists diff --git a/orchestrator/utils/orchestrator_functions.py b/orchestrator/utils/orchestrator_functions.py index b8e7eec..8b7c509 100644 --- a/orchestrator/utils/orchestrator_functions.py +++ b/orchestrator/utils/orchestrator_functions.py @@ -1,5 +1,7 @@ from typing import Any +from orchestrator.utils.converters import parse_frequency + def common_config(config: dict[str, Any]): """ @@ -7,15 +9,22 @@ def common_config(config: dict[str, Any]): Args: config (dict[str, Any]): Pipeline configuration containing: - - workflow_type (str): Type of workflow (e.g., 'scouter', 'predictions_batch') - - schedule_name (str): Name of the schedule - - frequency (str, optional): Frequency of execution (default: '1m') - - max_retry_policy (int, optional): Maximum retry attempts (default: 1) - - model_id (str): ID of the model - - models (dict): Model configuration containing 'name' field + - workflow_type (str): Type of workflow (e.g., 'scouter', 'predictions_batch', 'drift') + - schedule_name (str): Unique name identifier for this schedule + - model_id (str): MongoDB ID of the associated model + - model (dict): Model configuration containing: + - name (str): Human-readable name of the model + - model_config (dict, optional): Additional model-specific configuration + - frequency (str, optional): Execution frequency (default: '1m') + Format: '{number}{unit}' where unit is 's', 'm', 'h', or 'd' + - offset (str, optional): Schedule offset/delay (default: '0m') + - max_retry_policy (int, optional): Maximum retry attempts on failure (default: 1) + - execution_timeout_seconds (int, optional): Workflow execution timeout (default: 300) + - task_timeout_seconds (int, optional): Individual task timeout (default: 300) Returns: - dict[str, Any]: Common configuration dictionary with extracted parameters. + dict[str, Any]: Common configuration dictionary with standardized parameters + for Temporal workflow execution """ model = config['model'] return { @@ -41,13 +50,14 @@ def drift(config: dict[str, Any]): Args: config (dict[str, Any]): Pipeline configuration containing: - - interval_minutes (int, optional): Detection interval in minutes (default: 60) - - drift_metrics (list[str], optional): List of drift metrics to compute - (default: ['kolmogorov_smirnov', 'jensen_shannon', 'wasserstein']) + - interval_minutes (int, optional): Time window for data comparison in minutes (default: 60) + - drift_metrics (list[str], optional): Statistical metrics to compute (default: + ['kolmogorov_smirnov', 'jensen_shannon', 'wasserstein']) - Additional fields from common_config Returns: - dict[str, Any]: Drift configuration with workflow type set to 'drift' + dict[str, Any]: Drift detection configuration with source/target tables, + time interval, and metrics specifications. Results stored in 'drift_metrics' table """ return { **common_config(config), @@ -76,7 +86,9 @@ def simple_metrics(config: dict[str, Any]): - Additional fields from common_config Returns: - dict[str, Any]: Simple metrics configuration with workflow type set to 'simple_metrics' + dict[str, Any]: Simple metrics configuration with source tables (predictions and + actual data), target table, time interval, and metrics list. Results stored in + 'simple_metrics' table """ return { **common_config(config), @@ -95,12 +107,15 @@ def minimal_retrain(config: dict[str, Any]): Args: config (dict[str, Any]): Pipeline configuration containing: - - schedule_name (str): Name of the schedule - - query (str): SQL query for retraining + - query (str): SQL query to retrieve training data. Should return features + and target variable in expected format + - datetime_columns (list[str], optional): Column names to parse as datetime + for proper temporal handling (default: []) - Additional fields from common_config Returns: - dict[str, Any]: Minimal retrain configuration with workflow type set to 'minimal_retrain'. + dict[str, Any]: Minimal retrain configuration with SQL query, database settings, + and datetime column specifications. Retraining logs are stored in 'log_retrain' table """ return { **common_config(config), @@ -111,29 +126,63 @@ def minimal_retrain(config: dict[str, Any]): } +def base_scouter(config: dict[str, Any]): + """ + Build base scouter configuration shared by all scouter workflow types. + + Creates the foundational configuration for OPC data collection workflows, + including filter policies, database settings, and data retention parameters. + This configuration is extended by specific scouter implementations (OPC UA, PI Web API). + + Args: + config (dict[str, Any]): Pipeline configuration containing: + - filters (list[dict], optional): List of filter configurations with: + - filter_name (str): Name of the filter + - policy (str): Filter policy to apply + - tag_retention_minutes (int, optional): Tag retention time in minutes (default: 60) + - debug_data_package (bool, optional): Enable debug data package logging (default: False) + - fill_missing_tags (bool, optional): Fill missing tags with interpolation (default: False) + - Additional fields from common_config + + Returns: + dict[str, Any]: Base scouter configuration with filters, database settings, + and retention policies + """ + + filters = {} + for f in config.get('filters', []): + filters[f['filter_name']] = {'policy': f['policy']} + + return { + **common_config(config), + 'trigger_laborious': False, + 'filters': filters, + 'schema': 'sientia_data', + 'table_name': 'laborious_data', + 'retention_time': config.get('tag_retention_minutes', 60) * 60, + 'debug_data_package': config.get('debug_data_package', False), + 'fill_missing_tags': config.get('fill_missing_tags', False), + } + + def scouter(config: dict[str, Any]): """ Build scouter configuration from pipeline config. Args: config (dict[str, Any]): Pipeline configuration containing: - - filters (list[dict], optional): List of filter configurations + - schedule_name (str): Name of the schedule (used for topic generation) - read_tags (list[dict]): List of tag configurations with: - - filter_name (str): Name of the filter - - policy (str): Filter policy - - tag_name (str): Name of the tag - - aggr_func (str, optional): Aggregation function (default: 'lts') - - data_range (list[int], optional): Data range limits (default: [-100, 100]) - - tag_retention_minutes (int, optional): Tag retention time in minutes (default: 60) - - debug_data_package (bool, optional): Enable debug data package (default: False) - - Additional fields from common_config + - tag_name (str): Name of the tag to read + - aggr_func (str, optional): Aggregation function for data collection (default: 'lts') + Common values: 'lts' (last), 'avg' (average), 'min', 'max', 'sum' + - data_range (list[int], optional): Valid data range [min, max] (default: [-100, 100]) + - Additional fields from base_scouter Returns: - dict[str, Any]: Scouter configuration with topic, filters, tags, and retention settings. + dict[str, Any]: OPC UA scouter configuration with Kafka topic, tag mappings, + filters, and retention settings. Topic name follows pattern: 'raw_{schedule_name}' """ - filters = {} - for f in config.get('filters', []): - filters[f['filter_name']] = {'policy': f['policy']} tags = {} for tag in config['read_tags']: @@ -143,16 +192,65 @@ def scouter(config: dict[str, Any]): } return { - **common_config(config), + **base_scouter(config), 'topic': f'raw_{config["schedule_name"]}', - 'trigger_laborious': False, - 'filters': filters, - 'schema': 'sientia_data', - 'table_name': 'laborious_data', - 'retention_time': config.get('tag_retention_minutes', 60) * 60, 'model_tags': tags, - 'debug_data_package': config.get('debug_data_package', False), - 'fill_missing_tags': config.get('fill_missing_tags', False), + } + + +def pi_web_api_scouter(config: dict[str, Any]): + """ + Build PI Web API scouter configuration from pipeline config. + + Creates a data collection workflow configuration for OSIsoft PI servers using + the PI Web API REST interface. Configures tag mappings with WebIDs, aggregation + functions, and API query parameters including timeout management. + + Args: + config (dict[str, Any]): Pipeline configuration containing: + - read_tags (list[dict]): List of tag configurations with: + - tag_name (str): Name of the tag + - webid (str): PI Web API WebID for the tag + - aggr_func (str, optional): Aggregation function (default: 'lts') + - data_range (list[int], optional): Valid data range (default: [-100, 100]) + - pi_web_api_config (dict): PI Web API connection settings with: + - endpoint (str): PI Web API endpoint URL + - period (str, optional): Time period for data retrieval (default: '*-1d') + - max_count (int, optional): Maximum number of values to retrieve (default: 1) + - api_timeout (int, optional): API request timeout in seconds + - Additional fields from base_scouter + + Returns: + dict[str, Any]: PI Web API scouter configuration with tag mappings and query settings. + API timeout is automatically adjusted to not exceed workflow frequency. + """ + tags = {} + for tag, tag_config in config['read_tags'].items(): + tags[tag] = { + 'webid': tag_config['webid'], + 'aggr_func': tag_config.get('aggr_func', 'lts'), + 'data_range': tag_config.get('data_range', [-100, 100]), + } + + base_config = base_scouter(config) + + pi_web_api_config = config['pi_web_api_config'] + + config_timeout = pi_web_api_config.get('api_timeout', None) + frequency = parse_frequency(base_config['frequency']) + + if config_timeout is None or config_timeout > frequency: + config_timeout = frequency + + return { + **base_config, + 'model_tags': tags, + 'pi_web_api_query': { + 'endpoint': pi_web_api_config['endpoint'], + 'period': pi_web_api_config.get('period', '*-1d'), + 'max_count': pi_web_api_config.get('max_count', 1), + 'api_timeout': config_timeout, + }, } @@ -181,13 +279,25 @@ def overlap_filter_config(base_filter_config: dict[str, Any], config: list[dict[ def process_path_priority(path_priority: list[str]): """ - Process and normalize path priority list to ensure it contains the required priorities. + Process and normalize path priority list for filter policy execution order. + + Validates and normalizes the priority list used to determine the order in which + filter policies are evaluated in prediction workflows. Invalid priorities are + removed, missing required priorities are appended, and the list is truncated to + exactly 3 elements. + + Valid priorities define workflow behavior when filters are triggered: + - STOP: Halt workflow execution immediately + - CONTINUE: Proceed to next step despite filter trigger + - REPEAT: Retry the current step Args: - path_priority (list[str]): List of path priorities to process. + path_priority (list[str]): User-provided list of path priorities. May contain + invalid values or be incomplete. Returns: - list[str]: Normalized path priority list with exactly 3 elements: ["STOP", "CONTINUE", "REPEAT"]. + list[str]: Normalized path priority list with exactly 3 elements in user-specified + or default order. Default order when priorities are missing: ["STOP", "CONTINUE", "REPEAT"] """ for priority in path_priority[:]: if priority not in ['STOP', 'CONTINUE', 'REPEAT']: @@ -206,20 +316,25 @@ def predictions_batch(config: dict[str, Any]): Args: config (dict[str, Any]): Pipeline configuration containing: - - write_tags (list[dict]): List of tag configurations with: - - server_id (str): ID of the OPC server - - type (str): Tag type ('prediction' or 'confidence') - - addr (str): Tag address - - data_type (str, optional): Data type (default: 'float') - - path_priority (list[str], optional): List of path priorities (default: ["STOP", "CONTINUE", "REPEAT"]) - - input_filters (list[dict], optional): List of input filter configurations - - mlflow_transform_filters (list[dict], optional): List of MLflow transform filter configurations - - mlflow_predict_filters (list[dict], optional): List of MLflow predict filter configurations - - model_retention_minutes (int, optional): Model retention time in minutes (default: 60) + - query (str): SQL query to retrieve input data for predictions + - write_tags (list[dict]): List of OPC tag configurations for write-back with: + - server_id (str): ID of the target OPC server + - type (str): Tag type - 'prediction' (model output) or 'confidence' (prediction confidence) + - addr (str): OPC tag address/path + - data_type (str, optional): OPC data type (default: 'float') + - datetime_columns (list[str], optional): Column names to parse as datetime (default: []) + - path_priority (list[str], optional): Filter policy execution order (default: ["STOP", "CONTINUE", "REPEAT"]) + - input_filters (list[dict], optional): Input data validation filters + - mlflow_transform_filters (list[dict], optional): Transform stage filters + - mlflow_predict_filters (list[dict], optional): Prediction stage filters + - model_retention_minutes (int, optional): Data retention time in minutes (default: 60) + - save_transform (bool, optional): Save transformed data to database (default: True) + - predictions_storage_policy (str, optional): Prediction storage policy (default: 'lts:1') - Additional fields from common_config Returns: - dict[str, Any]: Predictions batch configuration with OPC output config, filters, and path priority. + dict[str, Any]: Complete predictions batch configuration with OPC output mappings, + multi-stage filters, SQL query, and retention policies """ tags: dict[str, Any] = {} for tag in config.get('write_tags', []): @@ -290,6 +405,9 @@ def gather_read_tags(pipelines: list[dict[str, Any]]) -> dict[str, Any]: # Get all read tags from pipelines for pipeline in pipelines: + if pipeline['workflow_type'] != 'scouter': + continue + for tag in pipeline.get('read_tags', []): tag_string = f'{tag["server_id"]}:{tag["tag_address"]}' if tag_string not in tags: diff --git a/tests/orchestrator/activities/test_formatters.py b/tests/orchestrator/activities/test_formatters.py index 51b0e08..f9df04c 100644 --- a/tests/orchestrator/activities/test_formatters.py +++ b/tests/orchestrator/activities/test_formatters.py @@ -21,6 +21,9 @@ def formatters(): formatters.send_notification = MagicMock() formatters.send_notification_async = AsyncMock() formatters.emit_metric = AsyncMock() + formatters.error = MagicMock() + formatters.info = MagicMock() + formatters.debug = MagicMock() return formatters @@ -35,31 +38,42 @@ metadata = { @mark.asyncio -@patch('orchestrator.activities.formatters.scouter', return_value={'test_scouter': 'test_scouter'}) -@patch( - 'orchestrator.activities.formatters.predictions_batch', - return_value={'test_predictions_batch': 'test_predictions_batch'}, -) -@patch( - 'orchestrator.activities.formatters.minimal_retrain', - return_value={'test_minimal_retrain': 'test_minimal_retrain'}, -) -@patch( - 'orchestrator.activities.formatters.drift', - return_value={'test_drift': 'test_drift'}, -) -@patch( - 'orchestrator.activities.formatters.simple_metrics', - return_value={'test_simple_metrics': 'test_simple_metrics'}, -) -async def test_process_schedules( - mock_simple_metrics, - mock_drift, - mock_minimal_retrain, - mock_predictions_batch, - mock_scouter, - formatters, -): +async def test_process_schedules(formatters): + mock_scouter = MagicMock(return_value={'test_scouter': 'test_scouter'}) + mock_predictions_batch = MagicMock( + return_value={'test_predictions_batch': 'test_predictions_batch'} + ) + mock_minimal_retrain = MagicMock(return_value={'test_minimal_retrain': 'test_minimal_retrain'}) + mock_drift = MagicMock(return_value={'test_drift': 'test_drift'}) + mock_simple_metrics = MagicMock(return_value={'test_simple_metrics': 'test_simple_metrics'}) + + mock_schedule_types = { + 'scouter': { + 'namespace': 'scouter', + 'function': mock_scouter, + }, + 'pi_web_api_scouter': { + 'namespace': 'scouter', + 'function': mock_scouter, + }, + 'predictions_batch': { + 'namespace': 'laborious', + 'function': mock_predictions_batch, + }, + 'minimal_retrain': { + 'namespace': 'laborious', + 'function': mock_minimal_retrain, + }, + 'drift': { + 'namespace': 'laborious', + 'function': mock_drift, + }, + 'simple_metrics': { + 'namespace': 'laborious', + 'function': mock_simple_metrics, + }, + } + input_data = { 'pipelines': [ { @@ -100,7 +114,8 @@ async def test_process_schedules( ] } - result = await formatters.process_schedules(input_data) + with patch('orchestrator.activities.formatters.schedule_types', mock_schedule_types): + result = await formatters.process_schedules(input_data) assert result == { 'scouter': { @@ -133,6 +148,79 @@ async def test_process_schedules( mock_simple_metrics.assert_called_once_with(input_data['pipelines'][4]) +@mark.asyncio +async def test_process_schedules_with_invalid_workflow_type(formatters): + """Test that process_schedules handles invalid workflow types correctly""" + mock_scouter = MagicMock(return_value={'test_scouter': 'test_scouter'}) + + mock_schedule_types = { + 'scouter': { + 'namespace': 'scouter', + 'function': mock_scouter, + }, + } + + pipelines = [ + { + 'schedule_name': 'test_schedule_name_valid', + 'workflow_type': 'scouter', + 'model_name': 'test_model_name', + 'model_id': 'test_model_id', + 'updated_at': '2021-01-01', + }, + { + 'schedule_name': 'test_schedule_name_invalid', + 'workflow_type': 'invalid_workflow_type', + 'model_name': 'test_model_name', + 'model_id': 'test_model_id', + 'updated_at': '2021-01-02', + }, + { + 'schedule_name': 'test_schedule_name_valid2', + 'workflow_type': 'scouter', + 'model_name': 'test_model_name', + 'model_id': 'test_model_id', + 'updated_at': '2021-01-03', + }, + ] + + input_data = { + 'pipelines': pipelines, + 'metadata': { + 'schedule_name': 'test_schedule', + 'workflow_name': 'test_workflow', + }, + } + + with patch('orchestrator.activities.formatters.schedule_types', mock_schedule_types): + result = await formatters.process_schedules(input_data) + + # Assert that error was called for invalid workflow type + formatters.error.assert_called_once_with( + 'Workflow type invalid_workflow_type not supported', metadata=input_data['metadata'] + ) + + # Assert that only valid pipelines were processed + assert result == { + 'scouter': { + 'test_schedule_name_valid': { + 'test_scouter': 'test_scouter', + 'updated_at': '2021-01-01', + }, + 'test_schedule_name_valid2': { + 'test_scouter': 'test_scouter', + 'updated_at': '2021-01-03', + }, + }, + 'laborious': {}, + } + + # Assert that the mock function was called only for valid pipelines + assert mock_scouter.call_count == 2 + mock_scouter.assert_any_call(pipelines[0]) + mock_scouter.assert_any_call(pipelines[2]) + + @mark.asyncio @patch( 'orchestrator.activities.formatters.gather_read_tags', diff --git a/tests/orchestrator/utils/test_orchestrator_functions.py b/tests/orchestrator/utils/test_orchestrator_functions.py index 4db42a8..805a21e 100644 --- a/tests/orchestrator/utils/test_orchestrator_functions.py +++ b/tests/orchestrator/utils/test_orchestrator_functions.py @@ -1,12 +1,14 @@ from unittest.mock import call, patch from orchestrator.utils.orchestrator_functions import ( + base_scouter, build_tag_config, common_config, drift, gather_read_tags, minimal_retrain, overlap_filter_config, + pi_web_api_scouter, predictions_batch, process_path_priority, scouter, @@ -219,7 +221,15 @@ def test_predictions_batch(mock_process_path_priority, mock_overlap_filter_confi [call({'EMPTY_DATA': {'policy': 'STOP', 'config': {}}}, config['input_filters'])] ) mock_overlap_filter_config.assert_has_calls( - [call({'API_ERROR': {'policy': 'STOP', 'config': {}}}, config['mlflow_transform_filters'])] + [ + call( + { + 'EMPTY_DATA': {'policy': 'STOP', 'config': {}}, + 'API_ERROR': {'policy': 'STOP', 'config': {}}, + }, + config['mlflow_transform_filters'], + ) + ] ) mock_overlap_filter_config.assert_has_calls( [call({'API_ERROR': {'policy': 'STOP', 'config': {}}}, config['mlflow_predict_filters'])] @@ -262,6 +272,7 @@ def test_predictions_batch(mock_process_path_priority, mock_overlap_filter_confi def test_gather_read_tags(): pipelines = [ { + 'workflow_type': 'scouter', 'schedule_name': 'test_schedule', 'read_tags': [ { @@ -272,6 +283,7 @@ def test_gather_read_tags(): ], }, { + 'workflow_type': 'scouter', 'schedule_name': 'test_schedule2', 'read_tags': [ { @@ -314,6 +326,103 @@ def test_gather_read_tags(): assert result == expected +def test_gather_read_tags_filters_non_scouter_workflows(): + """Test that gather_read_tags only processes scouter workflow types and ignores others""" + pipelines = [ + { + 'workflow_type': 'scouter', + 'schedule_name': 'test_scouter_schedule', + 'read_tags': [ + { + 'server_id': '1', + 'server_name': 'test_server_name', + 'tag_address': 'test_tag_address', + } + ], + }, + { + 'workflow_type': 'predictions_batch', + 'schedule_name': 'test_predictions_schedule', + 'read_tags': [ + { + 'server_id': '2', + 'server_name': 'test_server_name2', + 'tag_address': 'test_tag_address_predictions', + } + ], + }, + { + 'workflow_type': 'minimal_retrain', + 'schedule_name': 'test_retrain_schedule', + 'read_tags': [ + { + 'server_id': '3', + 'server_name': 'test_server_name3', + 'tag_address': 'test_tag_address_retrain', + } + ], + }, + { + 'workflow_type': 'pi_web_api_scouter', + 'schedule_name': 'test_pi_web_api_schedule', + 'read_tags': [ + { + 'server_id': '4', + 'server_name': 'test_server_name4', + 'tag_address': 'test_tag_address_pi_web_api', + } + ], + }, + { + 'workflow_type': 'drift', + 'schedule_name': 'test_drift_schedule', + 'read_tags': [ + { + 'server_id': '5', + 'server_name': 'test_server_name5', + 'tag_address': 'test_tag_address_drift', + } + ], + }, + { + 'workflow_type': 'scouter', + 'schedule_name': 'test_scouter_schedule2', + 'read_tags': [ + { + 'server_id': '1', + 'server_name': 'test_server_name', + 'tag_address': 'test_tag_address2', + } + ], + }, + ] + + result = gather_read_tags(pipelines) + + # Only scouter workflow types should be included + expected = { + '1:test_tag_address': { + 'server_id': '1', + 'server_name': 'test_server_name', + 'tag_address': 'test_tag_address', + 'topics': ['raw_test_scouter_schedule'], + }, + '1:test_tag_address2': { + 'server_id': '1', + 'server_name': 'test_server_name', + 'tag_address': 'test_tag_address2', + 'topics': ['raw_test_scouter_schedule2'], + }, + } + + assert result == expected + # Ensure non-scouter pipelines are not included + assert '2:test_tag_address_predictions' not in result + assert '3:test_tag_address_retrain' not in result + assert '4:test_tag_address_pi_web_api' not in result + assert '5:test_tag_address_drift' not in result + + def test_build_tag_config(): tags = [ { @@ -403,3 +512,203 @@ def test_build_tag_config(): } assert result == (expected, ['3']) + + +def test_base_scouter(): + config = { + 'workflow_type': 'scouter', + 'schedule_name': 'test_schedule', + 'model_id': 'test_model_id', + 'model': {'name': 'test_model_name', 'model_config': {'test_config': 'test_config'}}, + 'filters': [ + {'filter_name': 'test_filter_name', 'policy': 'test_policy'}, + {'filter_name': 'test_filter_name2', 'policy': 'test_policy2'}, + ], + 'tag_retention_minutes': 30, + 'debug_data_package': True, + 'fill_missing_tags': True, + } + result = base_scouter(config) + expected = { + 'workflow_type': 'scouter', + 'schedule_name': 'test_schedule', + 'frequency': '1m', + 'offset': '0m', + 'max_retry_policy': 1, + 'model_id': 'test_model_id', + 'model_name': 'test_model_name', + 'model_config': {'test_config': 'test_config'}, + 'trigger_laborious': False, + 'filters': { + 'test_filter_name': {'policy': 'test_policy'}, + 'test_filter_name2': {'policy': 'test_policy2'}, + }, + 'schema': 'sientia_data', + 'table_name': 'laborious_data', + 'retention_time': 30 * 60, + 'debug_data_package': True, + 'execution_timeout_seconds': 300, + 'task_timeout_seconds': 300, + 'fill_missing_tags': True, + } + assert result == expected + + +def test_pi_web_api_scouter(): + config = { + 'workflow_type': 'scouter', + 'schedule_name': 'test_schedule', + 'model_id': 'test_model_id', + 'model': {'name': 'test_model_name', 'model_config': {'test_config': 'test_config'}}, + 'filters': [{'filter_name': 'test_filter_name', 'policy': 'test_policy'}], + 'read_tags': { + 'test_tag_name': { + 'webid': 'test_webid', + 'aggr_func': 'test_aggr_func', + 'data_range': [1, 2], + } + }, + 'tag_retention_minutes': 10, + 'pi_web_api_config': { + 'endpoint': 'https://test-endpoint.com', + 'period': '*-2d', + 'max_count': 5, + 'api_timeout': 30, + }, + 'frequency': '1m', + } + result = pi_web_api_scouter(config) + expected = { + 'workflow_type': 'scouter', + 'schedule_name': 'test_schedule', + 'frequency': '1m', + 'offset': '0m', + 'max_retry_policy': 1, + 'model_id': 'test_model_id', + 'model_name': 'test_model_name', + 'model_config': {'test_config': 'test_config'}, + 'trigger_laborious': False, + 'filters': {'test_filter_name': {'policy': 'test_policy'}}, + 'schema': 'sientia_data', + 'table_name': 'laborious_data', + 'retention_time': 10 * 60, + 'model_tags': { + 'test_tag_name': { + 'webid': 'test_webid', + 'aggr_func': 'test_aggr_func', + 'data_range': [1, 2], + } + }, + 'debug_data_package': False, + 'execution_timeout_seconds': 300, + 'task_timeout_seconds': 300, + 'fill_missing_tags': False, + 'pi_web_api_query': { + 'endpoint': 'https://test-endpoint.com', + 'period': '*-2d', + 'max_count': 5, + 'api_timeout': 30, + }, + } + assert result == expected + + +def test_pi_web_api_scouter_with_timeout_greater_than_frequency(): + config = { + 'workflow_type': 'scouter', + 'schedule_name': 'test_schedule', + 'model_id': 'test_model_id', + 'model': {'name': 'test_model_name', 'model_config': {'test_config': 'test_config'}}, + 'filters': [], + 'read_tags': { + 'test_tag_name': { + 'webid': 'test_webid', + } + }, + 'tag_retention_minutes': 10, + 'pi_web_api_config': { + 'endpoint': 'https://test-endpoint.com', + 'api_timeout': 120, + }, + 'frequency': '1m', + } + result = pi_web_api_scouter(config) + expected = { + 'workflow_type': 'scouter', + 'schedule_name': 'test_schedule', + 'frequency': '1m', + 'offset': '0m', + 'max_retry_policy': 1, + 'model_id': 'test_model_id', + 'model_name': 'test_model_name', + 'model_config': {'test_config': 'test_config'}, + 'trigger_laborious': False, + 'filters': {}, + 'schema': 'sientia_data', + 'table_name': 'laborious_data', + 'retention_time': 10 * 60, + 'model_tags': { + 'test_tag_name': {'webid': 'test_webid', 'aggr_func': 'lts', 'data_range': [-100, 100]} + }, + 'debug_data_package': False, + 'execution_timeout_seconds': 300, + 'task_timeout_seconds': 300, + 'fill_missing_tags': False, + 'pi_web_api_query': { + 'endpoint': 'https://test-endpoint.com', + 'period': '*-1d', + 'max_count': 1, + 'api_timeout': 60, + }, + } + assert result == expected + + +def test_pi_web_api_scouter_with_no_timeout(): + config = { + 'workflow_type': 'scouter', + 'schedule_name': 'test_schedule', + 'model_id': 'test_model_id', + 'model': {'name': 'test_model_name', 'model_config': {'test_config': 'test_config'}}, + 'filters': [], + 'read_tags': { + 'test_tag_name': { + 'webid': 'test_webid', + } + }, + 'tag_retention_minutes': 10, + 'pi_web_api_config': { + 'endpoint': 'https://test-endpoint.com', + }, + 'frequency': '30s', + } + result = pi_web_api_scouter(config) + expected = { + 'workflow_type': 'scouter', + 'schedule_name': 'test_schedule', + 'frequency': '30s', + 'offset': '0m', + 'max_retry_policy': 1, + 'model_id': 'test_model_id', + 'model_name': 'test_model_name', + 'model_config': {'test_config': 'test_config'}, + 'trigger_laborious': False, + 'filters': {}, + 'schema': 'sientia_data', + 'table_name': 'laborious_data', + 'retention_time': 10 * 60, + 'model_tags': { + 'test_tag_name': {'webid': 'test_webid', 'aggr_func': 'lts', 'data_range': [-100, 100]} + }, + 'debug_data_package': False, + 'execution_timeout_seconds': 300, + 'task_timeout_seconds': 300, + 'fill_missing_tags': False, + 'pi_web_api_query': { + 'endpoint': 'https://test-endpoint.com', + 'period': '*-1d', + 'max_count': 1, + 'api_timeout': 30, + }, + } + assert result == expected diff --git a/values.yaml b/values.yaml index a40c8c4..fb697b8 100644 --- a/values.yaml +++ b/values.yaml @@ -151,7 +151,7 @@ env: - name: GITHUB_REPO_URL value: "git@github.com:Aignosi/sientia-dataops-orchestrator_temporal.git" - name: GITHUB_BRANCH - value: "feature/SIENTIAPDE-1273" + value: "feature/SIENTIAPDE-1445" - name: PYTHON_APP value: "orchestrator.worker.worker" @@ -182,17 +182,17 @@ env: - name: MONGODB_TTL_INDEX_HOURS value: "1" - - name: EMAIL_SENDER - value: "vitor.santos@aignosi.com.br" - - name: EMAIL_SENDER_PASSWORD - valueFrom: - secretKeyRef: - name: smtp-credentials - key: app_password - - name: EMAIL_SMTP_SERVER - value: "smtp.gmail.com" - - name: EMAIL_SMTP_PORT - value: "587" + # - name: EMAIL_SENDER + # value: "vitor.santos@aignosi.com.br" + # - name: EMAIL_SENDER_PASSWORD + # valueFrom: + # secretKeyRef: + # name: smtp-credentials + # key: app_password + # - name: EMAIL_SMTP_SERVER + # value: "smtp.gmail.com" + # - name: EMAIL_SMTP_PORT + # value: "587" # Application variables - name: POSTGRES_HOST