From c98469efd20e8e1679d7fc95189accf0f98cdff2 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 18 Dec 2025 10:55:29 -0300 Subject: [PATCH] SIENTIAPDE-1445 Update GITHUB_BRANCH to feature/SIENTIAPDE-1445 and enhance formatters with a new registry for workflow types, improving configuration management. Refactor process_schedules to utilize the new registry and add error handling for unsupported workflow types. Update tests to validate new functionality and ensure proper integration. --- orchestrator/activities/formatters.py | 118 +++++++++----- orchestrator/utils/orchestrator_functions.py | 153 +++++++++++++----- .../activities/test_formatters.py | 140 +++++++++++++--- .../utils/test_orchestrator_functions.py | 22 ++- values.yaml | 2 +- 5 files changed, 323 insertions(+), 112 deletions(-) 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/utils/orchestrator_functions.py b/orchestrator/utils/orchestrator_functions.py index 4f2b806..a2c5e16 100644 --- a/orchestrator/utils/orchestrator_functions.py +++ b/orchestrator/utils/orchestrator_functions.py @@ -9,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 { @@ -43,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), @@ -78,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), @@ -97,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), @@ -112,9 +125,28 @@ def minimal_retrain(config: dict[str, Any]): 'datetime_columns': config.get('datetime_columns', []), } + def base_scouter(config: dict[str, Any]): """ - Base scouter configuration. + 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 = {} @@ -139,22 +171,19 @@ def scouter(config: dict[str, Any]): 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}' """ - tags = {} for tag in config['read_tags']: tags[tag['tag_name']] = { @@ -168,9 +197,32 @@ def scouter(config: dict[str, Any]): 'model_tags': tags, } + 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 in config['read_tags']: @@ -198,7 +250,7 @@ def pi_web_api_scouter(config: dict[str, Any]): 'period': pi_web_api_config.get('period', '*-1d'), 'max_count': pi_web_api_config.get('max_count', 1), 'api_timeout': config_timeout, - } + }, } @@ -227,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']: @@ -252,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', []): 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 efcfe86..f225d45 100644 --- a/tests/orchestrator/utils/test_orchestrator_functions.py +++ b/tests/orchestrator/utils/test_orchestrator_functions.py @@ -486,7 +486,13 @@ def test_pi_web_api_scouter(): '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]}}, + '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, @@ -496,7 +502,7 @@ def test_pi_web_api_scouter(): 'period': '*-2d', 'max_count': 5, 'api_timeout': 30, - } + }, } assert result == expected @@ -536,7 +542,9 @@ def test_pi_web_api_scouter_with_timeout_greater_than_frequency(): '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]}}, + '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, @@ -546,7 +554,7 @@ def test_pi_web_api_scouter_with_timeout_greater_than_frequency(): 'period': '*-1d', 'max_count': 1, 'api_timeout': 60, - } + }, } assert result == expected @@ -585,7 +593,9 @@ def test_pi_web_api_scouter_with_no_timeout(): '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]}}, + '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, @@ -595,6 +605,6 @@ def test_pi_web_api_scouter_with_no_timeout(): 'period': '*-1d', 'max_count': 1, 'api_timeout': 30, - } + }, } assert result == expected diff --git a/values.yaml b/values.yaml index a40c8c4..ccad655 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"