From 3083c5edc4d093c3f1812fec227ffe7558af96a6 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Fri, 9 Jan 2026 09:56:50 -0300 Subject: [PATCH 1/9] SIENTIAPDE-1478 Update version in init_orchestration.ipynb to 3.11.14 and enhance README.md with detailed architecture principles, execution flows, and improved filtering mechanisms in workflows. Refactor orchestrator_functions.py for clarity in filter configuration and add support for PI Web API output configuration in predictions_batch. Update tests to reflect new configurations. --- README.md | 131 ++++++++++-------- init_orchestration.ipynb | 2 +- orchestrator/utils/orchestrator_functions.py | 66 ++++++--- .../utils/test_orchestrator_functions.py | 18 +++ 4 files changed, 133 insertions(+), 84 deletions(-) diff --git a/README.md b/README.md index 645679f..e84ac48 100644 --- a/README.md +++ b/README.md @@ -60,15 +60,15 @@ The SIENTIA DataOps Orchestrator uses a Temporal-based workflow architecture wit ### Architecture Principles #### 1. **Separation of Concerns** -- **Worker Layer**: Manages Temporal workers, task queues, and application lifecycle -- **Workflow Layer**: Orchestrates business logic and process coordination -- **Activity Layer**: Implements specific operations and external system interactions -- **Data Layer**: Handles data persistence, caching, and external service connections +- **Worker Layer**: Manages Temporal workers, task queues, Prometheus metrics, and application lifecycle +- **Workflow Layer**: Orchestrates business logic and process coordination with subworkflows for modularity +- **Activity Layer**: Implements specific operations and external system interactions with error handling +- **Data Layer**: Handles data persistence (MongoDB, PostgreSQL), caching (Redis), and external service connections #### 2. **Task Queue Isolation** -- **Orchestrator Queue**: Pipeline and resource management workflows -- **Alerts Queue**: Real-time error notification workflows -- **Reports Queue**: Scheduled reporting and summary workflows +- **Orchestrator Queue**: Pipeline and resource management workflows (orchestrator workflow) +- **Alerts Queue**: Real-time error notification workflows (alerts workflow, load_notification_package, process_notifications subworkflows) +- **Reports Queue**: Scheduled reporting and summary workflows (reports workflow, load_notification_package, process_notifications subworkflows) #### 3. **Fault Tolerance & Resilience** - **Automatic Retry Policies**: Configurable retry strategies for transient failures @@ -97,12 +97,11 @@ The **Orchestrator** workflow is the main coordination workflow that manages pip - **Infrastructure Management**: Creates, updates, and deletes workflow schedules **Execution Flow**: -1. **Configuration Loading**: Retrieves pipeline and OPC server configurations from MongoDB -2. **Resource Assessment**: Loads current OPC slots and active ingestors from Redis -3. **Schedule Processing**: Formats configurations for different workflow types -4. **Deployment Operations**: Creates, updates, or deletes Temporal schedules -5. **Resource Updates**: Updates OPC slots and MongoDB timestamps -6. **Reporting**: Generates comprehensive orchestration reports +1. **Parallel Data Loading**: Concurrently loads pipelines, OPC servers, orchestrated schedules, OPC slots, and active ingestors +2. **Parallel Processing**: Formats orchestrated schedules and processes new schedules and slots +3. **Parallel Config Creation**: Creates schedule and slot action configurations, normalizes schedules, and creates collections with TTL indexes +4. **Parallel Operations**: Executes slot deletions, slot updates, schedule deletions, schedule creations, and schedule updates concurrently +5. **Parallel Reports & Timestamps**: Generates orchestration reports and updates MongoDB timestamps for created/updated/deleted pipelines **Input Parameters**: ```json @@ -188,12 +187,13 @@ The **Alerts** workflow processes and sends real-time error notifications to con - **Persistent Monitoring**: Tracks and escalates persistent issues **Execution Flow**: -1. **Notification Loading**: Retrieves ERROR-level notifications from MongoDB -2. **Timestamp Filtering**: Applies incremental processing using Redis timestamps -3. **Group Filtering**: Filters notifications by user group configurations -4. **TTL Processing**: Checks notification cache to prevent duplicate alerts -5. **Email Generation**: Creates HTML email content for each group -6. **Delivery & Logging**: Sends emails and logs results to PostgreSQL +1. **Notification Loading**: Subworkflow loads ERROR-level notifications from MongoDB with timestamp filtering +2. **Configuration Loading**: Loads active receiver group configurations from MongoDB +3. **Alert Filtering**: Applies TTL-based filtering to identify core_alerts (new) and persistent_alerts (ongoing issues) +4. **Group Filtering**: Filters notifications by user group content policies and ignore lists +5. **Email Processing**: Subworkflow generates HTML emails and sends to receiver groups +6. **Cache Storage**: Stores sent notification cache in Redis with configurable TTL to prevent duplicates +7. **Audit Logging**: Logs delivery results to PostgreSQL for monitoring **Input Parameters**: ```json @@ -231,12 +231,11 @@ The **Reports** workflow generates and sends scheduled comprehensive reports to - **Audit Trail**: Complete logging of report delivery **Execution Flow**: -1. **Data Collection**: Loads all notifications from MongoDB (any level) -2. **Timestamp Processing**: Uses incremental loading with Redis timestamps -3. **Group Processing**: Applies user group filtering for report customization -4. **Report Generation**: Creates HTML reports with comprehensive summaries -5. **Distribution**: Sends reports to configured recipients -6. **Audit Logging**: Records delivery status in PostgreSQL +1. **Notification Loading**: Subworkflow loads all notifications from MongoDB (any level) with timestamp filtering +2. **Configuration Loading**: Loads active receiver group configurations from MongoDB +3. **Report Filtering**: Filters notifications by user group content policies (must include "reports") and ignore lists +4. **Email Processing**: Subworkflow generates HTML reports organized by notification level and model, then sends to receiver groups +5. **Audit Logging**: Logs delivery results to PostgreSQL for monitoring and tracking **Input Parameters**: ```json @@ -288,10 +287,10 @@ flowchart LR - `sending_configs` (list[dict]): Active receiver group configurations **Key Activities**: -- `get_last_data_timestamp`: Retrieves last processed timestamp from Redis -- `find_documents_in_mongodb`: Loads receiver group configurations -- `load_latest_data`: Loads notifications with timestamp filtering -- `put_last_data_timestamp`: Updates last processed timestamp +- `get_last_data_timestamp`: Retrieves last processed timestamp from Redis for incremental processing +- `find_documents_in_mongodb`: Loads active receiver group configurations from MongoDB +- `load_latest_data`: Loads notifications with timestamp filtering from notification_queue collection +- `put_last_data_timestamp`: Updates last processed timestamp in Redis with 5-hour TTL **Architecture**: @@ -343,10 +342,10 @@ flowchart LR - `log_report` (list[dict]): Detailed delivery status for each notification **Key Activities**: -- `build_email_html`: Generates HTML content using Jinja2 templates -- `send_email`: Delivers emails to receiver groups with error handling -- `format_log_report`: Formats delivery results for database storage -- `export_data_to_postgres`: Stores audit logs in PostgreSQL +- `build_email_html`: Generates HTML content using Jinja2 templates organized by notification level and model +- `send_email`: Delivers emails to receiver groups with attachment support and automatic SMTP reconnection +- `format_log_report`: Formats delivery results into DataFrame structure for database storage, aggregating by notification ID and trigger +- `export_data_to_postgres`: Stores audit logs in PostgreSQL with timestamp conversion **Architecture**: @@ -373,10 +372,11 @@ The orchestrator includes an advanced notification filtering system that prevent - **Ignore Lists**: Supports notification exclusion per group ### **Report Filtering (`filter_notification_reports`)** -- **Purpose**: Filters all notification levels for comprehensive reports +- **Purpose**: Filters all notification levels for comprehensive scheduled reports - **Comprehensive Coverage**: Includes ERROR, WARNING, INFO, and DEBUG levels -- **Group Customization**: Applies different content policies per receiver group -- **Scheduled Processing**: Designed for regular report generation +- **Group Customization**: Applies different content policies per receiver group with ignore list support +- **Scheduled Processing**: Designed for regular report generation without TTL-based duplicate prevention +- **Duplicate Prevention**: Prevents duplicate notifications within the same report using trigger and notification ID keys ### **Notification Caching (`store_notification_cache`)** - **Purpose**: Manages Redis-based notification cache for TTL enforcement @@ -389,32 +389,40 @@ The orchestrator includes an advanced notification filtering system that prevent #### **Worker (`orchestrator/worker/worker.py`)** - **Purpose**: Main application orchestrator managing Temporal workers and task queues - **Responsibilities**: - - Temporal client initialization and connection management + - Temporal client initialization and connection management with SDK metrics - Worker lifecycle management and graceful shutdown - - Task queue configuration (orchestrator, alerts, reports) - - Prometheus metrics server initialization + - Task queue configuration (orchestrator, alerts, reports) with dedicated workers + - Prometheus metrics server initialization on HTTP_METRICS_PORT + - Temporal SDK metrics server initialization on HTTP_SDK_METRICS_PORT - Notification handler setup and configuration - **Key Features**: - - Multi-queue worker management with automatic scaling - - Health check endpoints for Kubernetes liveness/readiness probes - - Graceful shutdown with cleanup procedures + - Multi-queue worker management with three dedicated workers (orchestrator, alerts, reports) + - Application health metrics (app_up gauge) for Kubernetes liveness/readiness probes + - Graceful shutdown with cleanup procedures for all connections - Comprehensive error handling and metrics collection + - Parallel worker execution using asyncio.gather #### **Activities (`orchestrator/activities/`)** -- **Activities**: Main activity orchestrator combining all operations -- **TemporalManager**: Temporal schedule CRUD operations across namespaces -- **SlotManager**: Redis-based OPC slot and cache management with notification filtering -- **MongoDB**: Document operations, aggregations, and TTL management -- **Email**: SMTP operations with HTML generation and attachment support -- **Formatters**: Configuration processing, slot distribution algorithms, and notification filtering for reports +- **Activities**: Main activity orchestrator combining all operations (TemporalManager, SlotManager, Formatters, MongoDB, Email, Postgres) +- **TemporalManager**: Temporal schedule CRUD operations across scouter and laborious namespaces with search attributes +- **SlotManager**: Redis-based OPC slot and cache management with notification filtering and TTL-based duplicate prevention +- **MongoDB**: Document operations, aggregations, timestamp management, and TTL index creation +- **Email**: SMTP operations with HTML generation, attachment support, and automatic reconnection handling +- **Formatters**: Configuration processing, slot distribution algorithms, notification filtering for reports, and schedule/slot orchestration reporting - **Postgres**: PostgreSQL operations for audit logging and data export (via sientia-dataops-library) #### **Utilities (`orchestrator/utils/`)** -- **Connectors Configuration**: Database and service configuration management -- **Email Builder**: HTML email template generation and formatting using Jinja2 templates -- **Orchestrator Functions**: Pipeline configuration transformation utilities for scouter, predictions_batch, minimal_retrain, drift, and simple_metrics workflows -- **Converters**: Data type conversion and validation utilities including frequency parsing -- **Templates**: HTML email templates for alerts and reports +- **Connectors Configuration**: Database and service configuration management from environment variables +- **Email Builder**: HTML email template generation and formatting using Jinja2 templates with support for alerts and reports +- **Orchestrator Functions**: Pipeline configuration transformation utilities supporting: + - `scouter`: OPC UA data collection workflows + - `pi_web_api_scouter`: PI Web API data collection workflows + - `predictions_batch`: ML prediction workflows with OPC write-back and multi-stage filtering + - `minimal_retrain`: Model retraining workflows with SQL queries + - `drift`: Data drift detection workflows + - `simple_metrics`: Model performance metrics computation workflows +- **Converters**: Data type conversion and validation utilities including frequency parsing for Temporal schedules +- **Templates**: HTML email templates for alerts and reports (email_template.html, general_template.html) ## ๐Ÿ“‹ Prerequisites @@ -621,18 +629,19 @@ Temporal input configuration sample: The Orchestrator system exposes comprehensive Prometheus metrics: ### Application Metrics -- `app_up`: Application health status (1=healthy, 0=unhealthy) -- `email_sent_count`: Email delivery operation count by group +- `app_up`: Application health status gauge (1=healthy, 0=unhealthy) labeled by pod_id +- `email_sent_count`: Email delivery counter labeled by pod_id, model_name, pipeline_name, and email_group ### Workflow Metrics -- Schedule creation, update, and deletion success rates -- Notification processing times and error rates -- Resource allocation and slot management metrics +- Schedule creation, update, and deletion success rates (via notification reports) +- Notification processing times and error rates (via PostgreSQL audit logs) +- Resource allocation and slot management metrics (via notification reports) +- Temporal SDK metrics exposed on HTTP_SDK_METRICS_PORT (default: 9091) ### Database Metrics -- MongoDB query performance and connection health -- Redis operation counts and response times -- PostgreSQL export operations and audit log metrics +- MongoDB query performance and connection health (via sientia-dataops-library) +- Redis operation counts and response times (via sientia-dataops-library) +- PostgreSQL export operations and audit log metrics (via sientia-dataops-library) ## ๐Ÿงช Testing diff --git a/init_orchestration.ipynb b/init_orchestration.ipynb index b63c2e0..6b97212 100644 --- a/init_orchestration.ipynb +++ b/init_orchestration.ipynb @@ -203,7 +203,7 @@ "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", - "version": "3.11.13" + "version": "3.11.14" } }, "nbformat": 4, diff --git a/orchestrator/utils/orchestrator_functions.py b/orchestrator/utils/orchestrator_functions.py index 8b7c509..0e6ad32 100644 --- a/orchestrator/utils/orchestrator_functions.py +++ b/orchestrator/utils/orchestrator_functions.py @@ -256,17 +256,23 @@ def pi_web_api_scouter(config: dict[str, Any]): def overlap_filter_config(base_filter_config: dict[str, Any], config: list[dict[str, Any]]): """ - Overlap filter configuration with base filter config. + Merge filter configurations with base filter configuration. + + Extends the base filter configuration dictionary by adding or overwriting + filters from the provided configuration list. Used in predictions_batch + workflows to combine default filters with user-defined custom filters. Args: - base_filter_config (dict[str, Any]): Base filter configuration to extend. - config (list[dict[str, Any]]): List of filter configurations to add, each containing: - - filter_name (str): Name of the filter - - policy (str): Filter policy - - config (dict, optional): Additional filter configuration + base_filter_config (dict[str, Any]): Base filter configuration dictionary to extend. + Each filter entry contains 'policy' and optionally 'config' keys. + config (list[dict[str, Any]]): List of filter configurations to merge, each containing: + - filter_name (str): Name of the filter to add or update + - policy (str): Filter policy (e.g., 'STOP', 'CONTINUE', 'REPEAT') + - config (dict, optional): Additional filter-specific configuration Returns: - dict[str, Any]: Extended filter configuration with new filters added. + dict[str, Any]: Extended filter configuration dictionary with merged filters. + Filters from config list overwrite or add to base_filter_config entries. """ for fil in config: base_filter_config[fil['filter_name']] = { @@ -330,11 +336,12 @@ def predictions_batch(config: dict[str, Any]): - 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') + - pi_web_api_output_config (dict, optional): PI Web API output configuration for write-back (default: {}) - Additional fields from common_config Returns: dict[str, Any]: Complete predictions batch configuration with OPC output mappings, - multi-stage filters, SQL query, and retention policies + PI Web API output configuration, multi-stage filters, SQL query, and retention policies """ tags: dict[str, Any] = {} for tag in config.get('write_tags', []): @@ -367,6 +374,7 @@ def predictions_batch(config: dict[str, Any]): 'transform_table_name': 'transformed_data', 'retention_time': config.get('model_retention_minutes', 60) * 60, 'opc_output_config': tags, + 'pi_web_api_output_config': config.get('pi_web_api_output_config', {}), 'input_filters': overlap_filter_config( {'EMPTY_DATA': {'policy': 'STOP', 'config': {}}}, config.get('input_filters', []) ), @@ -388,17 +396,24 @@ def predictions_batch(config: dict[str, Any]): def gather_read_tags(pipelines: list[dict[str, Any]]) -> dict[str, Any]: """ - Gather all read tags from input pipelines. + Gather all read tags from scouter pipeline configurations. - Collects all read tags from scouter pipelines and organizes them by - server_id and tag_address, tracking which topics each tag is associated with. + Collects all read tags from scouter-type pipelines and organizes them by + server_id and tag_address, tracking which Kafka topics each tag is associated with. + This function is used during slot configuration to aggregate tags across multiple + scouter pipelines for efficient OPC server slot allocation. Args: - pipelines (list[dict[str, Any]]): The pipeline configurations to process + pipelines (list[dict[str, Any]]): List of pipeline configurations to process. + Only pipelines with workflow_type 'scouter' are processed. Each scouter + pipeline should contain a 'read_tags' list with tag configurations. Returns: dict[str, Any]: Dictionary of read tags keyed by "server_id:tag_address", - each containing tag configuration and associated topics + where each entry contains: + - All original tag configuration fields + - topics (list[str]): List of Kafka topic names associated with this tag + (format: 'raw_{schedule_name}') """ tags = {} @@ -424,24 +439,31 @@ def build_tag_config( """ Build tag configuration for a specific slot and OPC server. - Organizes tags by OPC server and calculates the minimum subscription period + Organizes tags by OPC server name and calculates the minimum subscription period based on tag frequencies. Validates that all server IDs exist in the OPC - servers configuration. + servers configuration. The subscription period is set to half of the minimum + tag frequency to ensure efficient data collection. Args: tags (list[dict[str, Any]]): List of tag configurations containing: - server_id (str): ID of the OPC server - - tag_address (str): Address of the tag + - tag_address (str): Address/path of the OPC tag - frequency (int): Tag read frequency in milliseconds - opc_servers (dict[str, Any]): Dictionary of OPC server configurations + - Additional tag-specific configuration fields + opc_servers (dict[str, Any]): Dictionary of OPC server configurations keyed by server_id. + Each server configuration should contain: + - server_name (str): Human-readable server name + - url (str): OPC server URL + - uri (str): OPC server URI + - cert_path (str, optional): Certificate file path + - private_key_path (str, optional): Private key file path + - server_cert_path (str, optional): Server certificate file path Returns: tuple[dict[str, Any], list]: A tuple containing: - - Slot configuration dictionary organized by server name - - List of server IDs that were not found in opc_servers - - Raises: - ValueError: If the specified server_id is not found in opc_servers + - Slot configuration dictionary organized by server_name, where each server + contains connection details, tags dictionary, and subscription_period_ms + - List of server IDs (str) that were not found in opc_servers configuration """ slot_config = {} diff --git a/tests/orchestrator/utils/test_orchestrator_functions.py b/tests/orchestrator/utils/test_orchestrator_functions.py index 805a21e..a349a19 100644 --- a/tests/orchestrator/utils/test_orchestrator_functions.py +++ b/tests/orchestrator/utils/test_orchestrator_functions.py @@ -207,6 +207,15 @@ def test_predictions_batch(mock_process_path_priority, mock_overlap_filter_confi {'server_id': 'test_server_id', 'type': 'prediction', 'addr': 'test_addr'}, {'server_id': 'test_server_id', 'type': 'confidence', 'addr': 'test_addr'}, ], + 'pi_web_api_output_config': { + 'endpoint': 'test_endpoint', + 'prediction_tags': { + 'tag_1': 'webid_1', + }, + 'confidence_tags': { + 'tag_2': 'webid_2', + }, + }, 'input_filters': [{'filter_name': 'test_filter_name', 'policy': 'test_policy'}], 'mlflow_transform_filters': [{'filter_name': 'test_filter_name', 'policy': 'test_policy'}], 'mlflow_predict_filters': [{'filter_name': 'test_filter_name', 'policy': 'test_policy'}], @@ -257,6 +266,15 @@ def test_predictions_batch(mock_process_path_priority, mock_overlap_filter_confi 'confidence_tags': {'test_addr': {'data_type': 'float'}}, } }, + 'pi_web_api_output_config': { + 'endpoint': 'test_endpoint', + 'prediction_tags': { + 'tag_1': 'webid_1', + }, + 'confidence_tags': { + 'tag_2': 'webid_2', + }, + }, 'input_filters': {'test_filter_name': {'policy': 'test_policy', 'config': {}}}, 'mlflow_transform_filters': {'test_filter_name': {'policy': 'test_policy', 'config': {}}}, 'mlflow_predict_filters': {'test_filter_name': {'policy': 'test_policy', 'config': {}}}, From 4925b2b2bebfb1210075342685113c69b124dd38 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Fri, 9 Jan 2026 10:05:36 -0300 Subject: [PATCH 2/9] SIENTIAPDE-1478 Update sientia-dataops-library dependency to version 1.8.0 in requirements.txt --- requirements.txt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/requirements.txt b/requirements.txt index 315579b..9225226 100644 --- a/requirements.txt +++ b/requirements.txt @@ -4,5 +4,5 @@ sqlalchemy redis pymongo jinja2 -git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.7.1 +git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.8.0 prometheus-client From 9079003100e48e3c9f68fd370f2729938b9ea150 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Fri, 9 Jan 2026 12:35:41 -0300 Subject: [PATCH 3/9] SIENTIAPDE-1478 Refactor worker initialization by introducing prepare_worker function to streamline worker setup for orchestrator, alerts, and reports workflows, enhancing code clarity and maintainability. --- orchestrator/worker/prepare_worker.py | 72 +++++++++++++++++++++++++++ orchestrator/worker/worker.py | 30 ++++++----- 2 files changed, 89 insertions(+), 13 deletions(-) create mode 100644 orchestrator/worker/prepare_worker.py diff --git a/orchestrator/worker/prepare_worker.py b/orchestrator/worker/prepare_worker.py new file mode 100644 index 0000000..c8b68f2 --- /dev/null +++ b/orchestrator/worker/prepare_worker.py @@ -0,0 +1,72 @@ +import os +import re +from collections.abc import Sequence +from typing import Any + +from sientia_do.observability.logger import Logger +from temporalio.client import Client +from temporalio.worker import PollerBehaviorAutoscaling, Worker + +parameters = [ + ('MAX_CONCURRENT_WORKFLOW_TASKS', '200'), + ('MAX_CONCURRENT_ACTIVITIES', '200'), + ('MAX_CONCURRENT_LOCAL_ACTIVITIES', '200'), + ('MAX_CACHED_WORKFLOWS', '200'), + ('WORKFLOW_POLLER_BEHAVIUR_MINIMUM', '10'), + ('WORKFLOW_POLLER_BEHAVIUR_INITIAL', '100'), + ('WORKFLOW_POLLER_BEHAVIUR_MAXIMUM', '200'), + ('ACTIVITY_POLLER_BEHAVIUR_MINIMUM', '10'), + ('ACTIVITY_POLLER_BEHAVIUR_INITIAL', '100'), + ('ACTIVITY_POLLER_BEHAVIUR_MAXIMUM', '200'), +] + + +def camel_to_snake(text: str) -> str: + """Convert camelCase or PascalCase to snake_case.""" + text = re.sub('(.)([A-Z][a-z]+)', r'\1_\2', text) + text = re.sub('([a-z0-9])([A-Z])', r'\1_\2', text) + return text.lower() + + +def prepare_worker( + main_workflow: type, + other_workflows: Sequence[type], + activities: Sequence[Any], + temporal_client: Client, + logger: Logger, +) -> Worker: + main_workflow_name = main_workflow.__name__.upper() + + queue_name = f'{camel_to_snake(main_workflow.__name__)}-queue' + + local_workflow_parameters = {} + + for parameter in parameters: + local_workflow_parameters[parameter[0]] = int( + os.getenv(main_workflow_name + '_' + parameter[0], parameter[1]) + ) + + logger.info(f'Preparing worker for {main_workflow_name} with queue {queue_name}') + + return Worker( + temporal_client, + task_queue=queue_name, + workflows=[main_workflow, *other_workflows], + activities=[*activities], + max_concurrent_workflow_tasks=local_workflow_parameters['MAX_CONCURRENT_WORKFLOW_TASKS'], + max_concurrent_activities=local_workflow_parameters['MAX_CONCURRENT_ACTIVITIES'], + max_concurrent_local_activities=local_workflow_parameters[ + 'MAX_CONCURRENT_LOCAL_ACTIVITIES' + ], + max_cached_workflows=local_workflow_parameters['MAX_CACHED_WORKFLOWS'], + workflow_task_poller_behavior=PollerBehaviorAutoscaling( + minimum=local_workflow_parameters['WORKFLOW_POLLER_BEHAVIUR_MINIMUM'], + initial=local_workflow_parameters['WORKFLOW_POLLER_BEHAVIUR_INITIAL'], + maximum=local_workflow_parameters['WORKFLOW_POLLER_BEHAVIUR_MAXIMUM'], + ), + activity_task_poller_behavior=PollerBehaviorAutoscaling( + minimum=local_workflow_parameters['ACTIVITY_POLLER_BEHAVIUR_MINIMUM'], + initial=local_workflow_parameters['ACTIVITY_POLLER_BEHAVIUR_INITIAL'], + maximum=local_workflow_parameters['ACTIVITY_POLLER_BEHAVIUR_MAXIMUM'], + ), + ) diff --git a/orchestrator/worker/worker.py b/orchestrator/worker/worker.py index fe65d4f..8943b03 100644 --- a/orchestrator/worker/worker.py +++ b/orchestrator/worker/worker.py @@ -7,6 +7,8 @@ with workflow.unsafe.imports_passed_through(): import os import sys + from orchestrator.worker.prepare_worker import prepare_worker + from prometheus_client import start_http_server from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler from sientia_do.observability.logger import get_logger @@ -17,7 +19,6 @@ with workflow.unsafe.imports_passed_through(): build_email_config, build_mongodb_config, build_postgres_config, - # build_couchbase_config, build_redis_config, build_temporal_config, ) @@ -104,10 +105,10 @@ async def main(): logger.custom_info('Starting Workers...', metadata=metadata) workers = [ - Worker( - temporal_client, - task_queue='orchestrator-queue', - workflows=[Orchestrator], + prepare_worker( + temporal_client=temporal_client, + main_workflow=Orchestrator, + other_workflows=[], activities=[ # Redis activities.load_active_ingestors, @@ -137,11 +138,12 @@ async def main(): activities.report_slot_orchestration, activities.format_schedule_config, ], + logger=logger, ), - Worker( - temporal_client, - task_queue='alerts-queue', - workflows=[Alerts, LoadNotificationPackage, ProcessNotifications], + prepare_worker( + temporal_client=temporal_client, + main_workflow=Alerts, + other_workflows=[LoadNotificationPackage, ProcessNotifications], activities=[ # Load notifications activities.get_last_data_timestamp, @@ -158,11 +160,12 @@ async def main(): # Store notification cache activities.store_notification_cache, ], + logger=logger, ), - Worker( - temporal_client, - task_queue='reports-queue', - workflows=[Reports, LoadNotificationPackage, ProcessNotifications], + prepare_worker( + temporal_client=temporal_client, + main_workflow=Reports, + other_workflows=[LoadNotificationPackage, ProcessNotifications], activities=[ # Load notifications activities.get_last_data_timestamp, @@ -177,6 +180,7 @@ async def main(): activities.format_log_report, activities.export_data_to_postgres, ], + logger=logger, ), ] From a46eaa23674adea9170713144d6ea213ae85cd7c Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Fri, 9 Jan 2026 12:36:21 -0300 Subject: [PATCH 4/9] SIENTIAPDE-1478 Reintroduce prepare_worker import in worker.py to maintain consistency in worker setup, ensuring clarity and proper initialization for orchestrator workflows. --- orchestrator/worker/worker.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/orchestrator/worker/worker.py b/orchestrator/worker/worker.py index 8943b03..fbce72c 100644 --- a/orchestrator/worker/worker.py +++ b/orchestrator/worker/worker.py @@ -1,14 +1,11 @@ from temporalio import client, workflow from temporalio.runtime import PrometheusConfig, Runtime, TelemetryConfig -from temporalio.worker import Worker with workflow.unsafe.imports_passed_through(): import asyncio import os import sys - from orchestrator.worker.prepare_worker import prepare_worker - from prometheus_client import start_http_server from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler from sientia_do.observability.logger import get_logger @@ -22,6 +19,7 @@ with workflow.unsafe.imports_passed_through(): build_redis_config, build_temporal_config, ) + from orchestrator.worker.prepare_worker import prepare_worker from orchestrator.workflows.alerts import Alerts from orchestrator.workflows.orchestrator import Orchestrator from orchestrator.workflows.reports import Reports From 5c9a7948deee45d439f46b09b0982c7a6e591aa5 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Fri, 9 Jan 2026 14:21:23 -0300 Subject: [PATCH 5/9] SIENTIAPDE-1478 Update init_orchestration.ipynb to reset execution counts and handle ScheduleAlreadyRunningError, ensuring proper error management during schedule creation. Modify values.yaml to change GITHUB_BRANCH to feature/SIENTIAPDE-1478 for alignment with ongoing feature development. --- init_orchestration.ipynb | 57 ++++++++++++++++++++++++++++------------ values.yaml | 2 +- 2 files changed, 41 insertions(+), 18 deletions(-) diff --git a/init_orchestration.ipynb b/init_orchestration.ipynb index 6b97212..a36f94d 100644 --- a/init_orchestration.ipynb +++ b/init_orchestration.ipynb @@ -2,7 +2,7 @@ "cells": [ { "cell_type": "code", - "execution_count": 5, + "execution_count": 1, "id": "3f8b77a4", "metadata": {}, "outputs": [], @@ -23,19 +23,33 @@ }, { "cell_type": "code", - "execution_count": 6, + "execution_count": null, "id": "d9d2a242", "metadata": {}, "outputs": [ { - "data": { - "text/plain": [ - "" - ] - }, - "execution_count": 6, - "metadata": {}, - "output_type": "execute_result" + "ename": "ScheduleAlreadyRunningError", + "evalue": "Schedule already running", + "output_type": "error", + "traceback": [ + "\u001b[31m---------------------------------------------------------------------------\u001b[39m", + "\u001b[31mRPCError\u001b[39m Traceback (most recent call last)", + "\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-orchestrator_temporal/venv/lib/python3.11/site-packages/temporalio/service.py:1243\u001b[39m, in \u001b[36m_BridgeServiceClient._rpc_call\u001b[39m\u001b[34m(self, rpc, req, resp_type, service, retry, metadata, timeout)\u001b[39m\n\u001b[32m 1242\u001b[39m client = \u001b[38;5;28;01mawait\u001b[39;00m \u001b[38;5;28mself\u001b[39m._connected_client()\n\u001b[32m-> \u001b[39m\u001b[32m1243\u001b[39m resp = \u001b[38;5;28;01mawait\u001b[39;00m client.call(\n\u001b[32m 1244\u001b[39m service=service,\n\u001b[32m 1245\u001b[39m rpc=rpc,\n\u001b[32m 1246\u001b[39m req=req,\n\u001b[32m 1247\u001b[39m resp_type=resp_type,\n\u001b[32m 1248\u001b[39m retry=retry,\n\u001b[32m 1249\u001b[39m metadata=metadata,\n\u001b[32m 1250\u001b[39m timeout=timeout,\n\u001b[32m 1251\u001b[39m )\n\u001b[32m 1252\u001b[39m \u001b[38;5;28;01mif\u001b[39;00m LOG_PROTOS:\n", + "\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-orchestrator_temporal/venv/lib/python3.11/site-packages/temporalio/bridge/client.py:151\u001b[39m, in \u001b[36mClient.call\u001b[39m\u001b[34m(self, service, rpc, req, resp_type, retry, metadata, timeout)\u001b[39m\n\u001b[32m 150\u001b[39m resp = resp_type()\n\u001b[32m--> \u001b[39m\u001b[32m151\u001b[39m resp.ParseFromString(\u001b[38;5;28;01mawait\u001b[39;00m resp_fut)\n\u001b[32m 152\u001b[39m \u001b[38;5;28;01mreturn\u001b[39;00m resp\n", + "\u001b[31mRPCError\u001b[39m: (6, 'Workflow execution is already running. WorkflowId: temporal-sys-scheduler:orchestrator, RunId: bd85451a-3e3e-4070-988a-f33c423a8db6.', b'\\x08\\x06\\x12\\x84\\x01Workflow execution is already running. WorkflowId: temporal-sys-scheduler:orchestrator, RunId: bd85451a-3e3e-4070-988a-f33c423a8db6.\\x1a\\xa7\\x01\\nWtype.googleapis.com/temporal.api.errordetails.v1.WorkflowExecutionAlreadyStartedFailure\\x12L\\n$be7bbffb-a83f-45b4-a8df-5ad197cc152b\\x12$bd85451a-3e3e-4070-988a-f33c423a8db6')", + "\nDuring handling of the above exception, another exception occurred:\n", + "\u001b[31mRPCError\u001b[39m Traceback (most recent call last)", + "\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-orchestrator_temporal/venv/lib/python3.11/site-packages/temporalio/client.py:6430\u001b[39m, in \u001b[36m_ClientImpl.create_schedule\u001b[39m\u001b[34m(self, input)\u001b[39m\n\u001b[32m 6427\u001b[39m temporalio.converter.encode_search_attributes(\n\u001b[32m 6428\u001b[39m \u001b[38;5;28minput\u001b[39m.search_attributes, request.search_attributes\n\u001b[32m 6429\u001b[39m )\n\u001b[32m-> \u001b[39m\u001b[32m6430\u001b[39m \u001b[38;5;28;01mawait\u001b[39;00m \u001b[38;5;28mself\u001b[39m._client.workflow_service.create_schedule(\n\u001b[32m 6431\u001b[39m request,\n\u001b[32m 6432\u001b[39m retry=\u001b[38;5;28;01mTrue\u001b[39;00m,\n\u001b[32m 6433\u001b[39m metadata=\u001b[38;5;28minput\u001b[39m.rpc_metadata,\n\u001b[32m 6434\u001b[39m timeout=\u001b[38;5;28minput\u001b[39m.rpc_timeout,\n\u001b[32m 6435\u001b[39m )\n\u001b[32m 6436\u001b[39m \u001b[38;5;28;01mexcept\u001b[39;00m RPCError \u001b[38;5;28;01mas\u001b[39;00m err:\n", + "\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-orchestrator_temporal/venv/lib/python3.11/site-packages/temporalio/service.py:1170\u001b[39m, in \u001b[36mServiceCall.__call__\u001b[39m\u001b[34m(self, req, retry, metadata, timeout)\u001b[39m\n\u001b[32m 1155\u001b[39m \u001b[38;5;250m\u001b[39m\u001b[33;03m\"\"\"Invoke underlying client with the given request.\u001b[39;00m\n\u001b[32m 1156\u001b[39m \n\u001b[32m 1157\u001b[39m \u001b[33;03mArgs:\u001b[39;00m\n\u001b[32m (...)\u001b[39m\u001b[32m 1168\u001b[39m \u001b[33;03m RPCError: Any RPC error that occurs during the call.\u001b[39;00m\n\u001b[32m 1169\u001b[39m \u001b[33;03m\"\"\"\u001b[39;00m\n\u001b[32m-> \u001b[39m\u001b[32m1170\u001b[39m \u001b[38;5;28;01mreturn\u001b[39;00m \u001b[38;5;28;01mawait\u001b[39;00m \u001b[38;5;28mself\u001b[39m.service_client._rpc_call(\n\u001b[32m 1171\u001b[39m \u001b[38;5;28mself\u001b[39m.name,\n\u001b[32m 1172\u001b[39m req,\n\u001b[32m 1173\u001b[39m \u001b[38;5;28mself\u001b[39m.resp_type,\n\u001b[32m 1174\u001b[39m service=\u001b[38;5;28mself\u001b[39m.service,\n\u001b[32m 1175\u001b[39m retry=retry,\n\u001b[32m 1176\u001b[39m metadata=metadata,\n\u001b[32m 1177\u001b[39m timeout=timeout,\n\u001b[32m 1178\u001b[39m )\n", + "\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-orchestrator_temporal/venv/lib/python3.11/site-packages/temporalio/service.py:1258\u001b[39m, in \u001b[36m_BridgeServiceClient._rpc_call\u001b[39m\u001b[34m(self, rpc, req, resp_type, service, retry, metadata, timeout)\u001b[39m\n\u001b[32m 1257\u001b[39m status, message, details = err.args\n\u001b[32m-> \u001b[39m\u001b[32m1258\u001b[39m \u001b[38;5;28;01mraise\u001b[39;00m RPCError(message, RPCStatusCode(status), details)\n", + "\u001b[31mRPCError\u001b[39m: Workflow execution is already running. WorkflowId: temporal-sys-scheduler:orchestrator, RunId: bd85451a-3e3e-4070-988a-f33c423a8db6.", + "\nDuring handling of the above exception, another exception occurred:\n", + "\u001b[31mScheduleAlreadyRunningError\u001b[39m Traceback (most recent call last)", + "\u001b[36mCell\u001b[39m\u001b[36m \u001b[39m\u001b[32mIn[3]\u001b[39m\u001b[32m, line 13\u001b[39m\n\u001b[32m 2\u001b[39m \u001b[38;5;28;01mfrom\u001b[39;00m\u001b[38;5;250m \u001b[39m\u001b[34;01mtemporalio\u001b[39;00m\u001b[34;01m.\u001b[39;00m\u001b[34;01mclient\u001b[39;00m\u001b[38;5;250m \u001b[39m\u001b[38;5;28;01mimport\u001b[39;00m (\n\u001b[32m 3\u001b[39m Schedule,\n\u001b[32m 4\u001b[39m ScheduleActionStartWorkflow,\n\u001b[32m 5\u001b[39m ScheduleIntervalSpec,\n\u001b[32m 6\u001b[39m ScheduleSpec,\n\u001b[32m 7\u001b[39m )\n\u001b[32m 9\u001b[39m \u001b[38;5;66;03m# temporal operator search-attribute create --namespace scouter --name model_id --type Text && temporal operator search-attribute create --namespace scouter --name orchestrated --type Text && temporal operator search-attribute create --namespace scouter --name model_name --type Text && temporal operator search-attribute create --namespace laborious --name model_id --type Text && temporal operator search-attribute create --namespace laborious --name orchestrated --type Text && temporal operator search-attribute create --namespace laborious --name model_name --type Text\u001b[39;00m\n\u001b[32m---> \u001b[39m\u001b[32m13\u001b[39m \u001b[38;5;28;01mawait\u001b[39;00m temporal_client.create_schedule(\n\u001b[32m 14\u001b[39m \u001b[33m\"\u001b[39m\u001b[33morchestrator\u001b[39m\u001b[33m\"\u001b[39m,\n\u001b[32m 15\u001b[39m Schedule(\n\u001b[32m 16\u001b[39m action=ScheduleActionStartWorkflow(\n\u001b[32m 17\u001b[39m \u001b[33m'\u001b[39m\u001b[33morchestrator\u001b[39m\u001b[33m'\u001b[39m,\n\u001b[32m 18\u001b[39m {\n\u001b[32m 19\u001b[39m \u001b[33m\"\u001b[39m\u001b[33mschedule_name\u001b[39m\u001b[33m\"\u001b[39m: \u001b[33m\"\u001b[39m\u001b[33morchestrator\u001b[39m\u001b[33m\"\u001b[39m,\n\u001b[32m 20\u001b[39m \u001b[33m\"\u001b[39m\u001b[33mpipelines_query\u001b[39m\u001b[33m\"\u001b[39m: {\n\u001b[32m 21\u001b[39m \u001b[33m\"\u001b[39m\u001b[33mcollection\u001b[39m\u001b[33m\"\u001b[39m: \u001b[33m\"\u001b[39m\u001b[33mpipelines\u001b[39m\u001b[33m\"\u001b[39m,\n\u001b[32m 22\u001b[39m \u001b[33m\"\u001b[39m\u001b[33maggregation\u001b[39m\u001b[33m\"\u001b[39m: [\n\u001b[32m 23\u001b[39m {\n\u001b[32m 24\u001b[39m \u001b[33m\"\u001b[39m\u001b[33m$lookup\u001b[39m\u001b[33m\"\u001b[39m: {\n\u001b[32m 25\u001b[39m \u001b[33m\"\u001b[39m\u001b[33mfrom\u001b[39m\u001b[33m\"\u001b[39m: \u001b[33m\"\u001b[39m\u001b[33mmodels\u001b[39m\u001b[33m\"\u001b[39m,\n\u001b[32m 26\u001b[39m \u001b[33m\"\u001b[39m\u001b[33mlocalField\u001b[39m\u001b[33m\"\u001b[39m: \u001b[33m\"\u001b[39m\u001b[33mmodel_id\u001b[39m\u001b[33m\"\u001b[39m,\n\u001b[32m 27\u001b[39m \u001b[33m\"\u001b[39m\u001b[33mforeignField\u001b[39m\u001b[33m\"\u001b[39m: \u001b[33m\"\u001b[39m\u001b[33mid\u001b[39m\u001b[33m\"\u001b[39m,\n\u001b[32m 28\u001b[39m \u001b[33m\"\u001b[39m\u001b[33mas\u001b[39m\u001b[33m\"\u001b[39m: \u001b[33m\"\u001b[39m\u001b[33mmodel\u001b[39m\u001b[33m\"\u001b[39m\n\u001b[32m 29\u001b[39m }\n\u001b[32m 30\u001b[39m },\n\u001b[32m 31\u001b[39m {\n\u001b[32m 32\u001b[39m \u001b[33m\"\u001b[39m\u001b[33m$unwind\u001b[39m\u001b[33m\"\u001b[39m: \u001b[33m\"\u001b[39m\u001b[33m$model\u001b[39m\u001b[33m\"\u001b[39m\n\u001b[32m 33\u001b[39m },\n\u001b[32m 34\u001b[39m {\n\u001b[32m 35\u001b[39m \u001b[33m\"\u001b[39m\u001b[33m$match\u001b[39m\u001b[33m\"\u001b[39m: {\n\u001b[32m 36\u001b[39m \u001b[33m\"\u001b[39m\u001b[33mactive\u001b[39m\u001b[33m\"\u001b[39m: \u001b[38;5;28;01mTrue\u001b[39;00m\n\u001b[32m 37\u001b[39m }\n\u001b[32m 38\u001b[39m },\n\u001b[32m 39\u001b[39m {\n\u001b[32m 40\u001b[39m \u001b[33m\"\u001b[39m\u001b[33m$match\u001b[39m\u001b[33m\"\u001b[39m: {\n\u001b[32m 41\u001b[39m \u001b[33m\"\u001b[39m\u001b[33mmodel.active\u001b[39m\u001b[33m\"\u001b[39m: \u001b[38;5;28;01mTrue\u001b[39;00m\n\u001b[32m 42\u001b[39m }\n\u001b[32m 43\u001b[39m }\n\u001b[32m 44\u001b[39m ]\n\u001b[32m 45\u001b[39m },\n\u001b[32m 46\u001b[39m \u001b[33m\"\u001b[39m\u001b[33mopc_servers_query\u001b[39m\u001b[33m\"\u001b[39m: {\n\u001b[32m 47\u001b[39m \u001b[33m\"\u001b[39m\u001b[33mcollection\u001b[39m\u001b[33m\"\u001b[39m: \u001b[33m\"\u001b[39m\u001b[33mopc-servers\u001b[39m\u001b[33m\"\u001b[39m,\n\u001b[32m 48\u001b[39m \u001b[33m\"\u001b[39m\u001b[33mfilters\u001b[39m\u001b[33m\"\u001b[39m: {\n\u001b[32m 49\u001b[39m \n\u001b[32m 50\u001b[39m }\n\u001b[32m 51\u001b[39m }\n\u001b[32m 52\u001b[39m },\n\u001b[32m 53\u001b[39m \u001b[38;5;28mid\u001b[39m=\u001b[33m\"\u001b[39m\u001b[33morchestrator\u001b[39m\u001b[33m\"\u001b[39m,\n\u001b[32m 54\u001b[39m task_queue=\u001b[33m\"\u001b[39m\u001b[33morchestrator-queue\u001b[39m\u001b[33m\"\u001b[39m,\n\u001b[32m 55\u001b[39m execution_timeout=timedelta(minutes=\u001b[32m600\u001b[39m)\n\u001b[32m 56\u001b[39m ),\n\u001b[32m 57\u001b[39m spec=ScheduleSpec(\n\u001b[32m 58\u001b[39m intervals=[ScheduleIntervalSpec(every=timedelta(minutes=\u001b[32m60\u001b[39m))]\n\u001b[32m 59\u001b[39m )\n\u001b[32m 60\u001b[39m )\n\u001b[32m 61\u001b[39m )\n", + "\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-orchestrator_temporal/venv/lib/python3.11/site-packages/temporalio/client.py:1308\u001b[39m, in \u001b[36mClient.create_schedule\u001b[39m\u001b[34m(self, id, schedule, trigger_immediately, backfill, memo, search_attributes, static_summary, static_details, rpc_metadata, rpc_timeout)\u001b[39m\n\u001b[32m 1275\u001b[39m \u001b[38;5;250m\u001b[39m\u001b[33;03m\"\"\"Create a schedule and return its handle.\u001b[39;00m\n\u001b[32m 1276\u001b[39m \n\u001b[32m 1277\u001b[39m \u001b[33;03mArgs:\u001b[39;00m\n\u001b[32m (...)\u001b[39m\u001b[32m 1305\u001b[39m \u001b[33;03m running.\u001b[39;00m\n\u001b[32m 1306\u001b[39m \u001b[33;03m\"\"\"\u001b[39;00m\n\u001b[32m 1307\u001b[39m temporalio.common._warn_on_deprecated_search_attributes(search_attributes)\n\u001b[32m-> \u001b[39m\u001b[32m1308\u001b[39m \u001b[38;5;28;01mreturn\u001b[39;00m \u001b[38;5;28;01mawait\u001b[39;00m \u001b[38;5;28mself\u001b[39m._impl.create_schedule(\n\u001b[32m 1309\u001b[39m CreateScheduleInput(\n\u001b[32m 1310\u001b[39m \u001b[38;5;28mid\u001b[39m=\u001b[38;5;28mid\u001b[39m,\n\u001b[32m 1311\u001b[39m schedule=schedule,\n\u001b[32m 1312\u001b[39m trigger_immediately=trigger_immediately,\n\u001b[32m 1313\u001b[39m backfill=backfill,\n\u001b[32m 1314\u001b[39m memo=memo,\n\u001b[32m 1315\u001b[39m search_attributes=search_attributes,\n\u001b[32m 1316\u001b[39m rpc_metadata=rpc_metadata,\n\u001b[32m 1317\u001b[39m rpc_timeout=rpc_timeout,\n\u001b[32m 1318\u001b[39m )\n\u001b[32m 1319\u001b[39m )\n", + "\u001b[36mFile \u001b[39m\u001b[32m~/Documents/projects/sientia/sientia-dataops-orchestrator_temporal/venv/lib/python3.11/site-packages/temporalio/client.py:6445\u001b[39m, in \u001b[36m_ClientImpl.create_schedule\u001b[39m\u001b[34m(self, input)\u001b[39m\n\u001b[32m 6437\u001b[39m already_started = (\n\u001b[32m 6438\u001b[39m err.status == RPCStatusCode.ALREADY_EXISTS\n\u001b[32m 6439\u001b[39m \u001b[38;5;129;01mand\u001b[39;00m err.grpc_status.details\n\u001b[32m (...)\u001b[39m\u001b[32m 6442\u001b[39m )\n\u001b[32m 6443\u001b[39m )\n\u001b[32m 6444\u001b[39m \u001b[38;5;28;01mif\u001b[39;00m already_started:\n\u001b[32m-> \u001b[39m\u001b[32m6445\u001b[39m \u001b[38;5;28;01mraise\u001b[39;00m ScheduleAlreadyRunningError()\n\u001b[32m 6446\u001b[39m \u001b[38;5;28;01mraise\u001b[39;00m\n\u001b[32m 6447\u001b[39m \u001b[38;5;28;01mreturn\u001b[39;00m ScheduleHandle(\u001b[38;5;28mself\u001b[39m._client, \u001b[38;5;28minput\u001b[39m.id)\n", + "\u001b[31mScheduleAlreadyRunningError\u001b[39m: Schedule already running" + ] } ], "source": [ @@ -45,7 +59,16 @@ " ScheduleActionStartWorkflow,\n", " ScheduleIntervalSpec,\n", " ScheduleSpec,\n", - ")\n", + ")\n" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "id": "c4703854", + "metadata": {}, + "outputs": [], + "source": [ "\n", "# temporal operator search-attribute create --namespace scouter --name model_id --type Text && temporal operator search-attribute create --namespace scouter --name orchestrated --type Text && temporal operator search-attribute create --namespace scouter --name model_name --type Text && temporal operator search-attribute create --namespace laborious --name model_id --type Text && temporal operator search-attribute create --namespace laborious --name orchestrated --type Text && temporal operator search-attribute create --namespace laborious --name model_name --type Text\n", "\n", @@ -104,17 +127,17 @@ }, { "cell_type": "code", - "execution_count": 8, + "execution_count": 4, "id": "6ea0f616", "metadata": {}, "outputs": [ { "data": { "text/plain": [ - "" + "" ] }, - "execution_count": 8, + "execution_count": 4, "metadata": {}, "output_type": "execute_result" } @@ -143,17 +166,17 @@ }, { "cell_type": "code", - "execution_count": 7, + "execution_count": 5, "id": "05c46ec8", "metadata": {}, "outputs": [ { "data": { "text/plain": [ - "" + "" ] }, - "execution_count": 7, + "execution_count": 5, "metadata": {}, "output_type": "execute_result" } diff --git a/values.yaml b/values.yaml index e6a0fb7..d9d40bb 100644 --- a/values.yaml +++ b/values.yaml @@ -164,7 +164,7 @@ env: - name: GITHUB_REPO_URL value: "git@github.com:Aignosi/sientia-dataops-orchestrator_temporal.git" - name: GITHUB_BRANCH - value: "fix/SIENTIAPDE-1461" + value: "feature/SIENTIAPDE-1478" - name: PYTHON_APP value: "orchestrator.worker.worker" From ba73bb7e9dacf63a1a37c54b855475e2aa0445a3 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Thu, 15 Jan 2026 11:08:48 -0300 Subject: [PATCH 6/9] SIENTIAPDE-1478 Add xgboost functions to schedule_types in formatters.py for enhanced orchestration capabilities --- orchestrator/activities/formatters.py | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/orchestrator/activities/formatters.py b/orchestrator/activities/formatters.py index 636d8a9..4f741c8 100644 --- a/orchestrator/activities/formatters.py +++ b/orchestrator/activities/formatters.py @@ -58,10 +58,18 @@ schedule_types: dict[str, ScheduleType] = { 'namespace': 'laborious', 'function': predictions_batch, }, + 'xgboost_predictions_batch': { + 'namespace': 'laborious', + 'function': predictions_batch, + }, 'minimal_retrain': { 'namespace': 'laborious', 'function': minimal_retrain, }, + 'xgboost_minimal_retrain': { + 'namespace': 'laborious', + 'function': minimal_retrain, + }, 'drift': { 'namespace': 'laborious', 'function': drift, From c9e3a15515e48591f55b81c1e362dd690b349dad Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Mon, 19 Jan 2026 10:14:36 -0300 Subject: [PATCH 7/9] SIENTIAPDE-1478 Update sonar-project.properties to broaden coverage exclusions for worker files, enhancing code quality analysis. --- sonar-project.properties | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sonar-project.properties b/sonar-project.properties index b6ee20e..92e208d 100644 --- a/sonar-project.properties +++ b/sonar-project.properties @@ -8,4 +8,4 @@ sonar.python.coverage.reportPaths=coverage.xml sonar.python.xunit.reportPath=pytest.xml sonar.python.version=3.11 sonar.projectVersion=1.0.0 -sonar.coverage.exclusions=orchestrator/worker/worker.py +sonar.coverage.exclusions=orchestrator/worker/* From 7f8105dd1e4a822973321c8aad77b2369d3d4ed6 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Wed, 21 Jan 2026 15:42:36 -0300 Subject: [PATCH 8/9] SIENTIAPDE-1478 Fix spelling errors in poller behavior parameters in prepare_worker.py for consistency and clarity. --- orchestrator/worker/prepare_worker.py | 24 ++++++++++++------------ 1 file changed, 12 insertions(+), 12 deletions(-) diff --git a/orchestrator/worker/prepare_worker.py b/orchestrator/worker/prepare_worker.py index c8b68f2..07ae36f 100644 --- a/orchestrator/worker/prepare_worker.py +++ b/orchestrator/worker/prepare_worker.py @@ -12,12 +12,12 @@ parameters = [ ('MAX_CONCURRENT_ACTIVITIES', '200'), ('MAX_CONCURRENT_LOCAL_ACTIVITIES', '200'), ('MAX_CACHED_WORKFLOWS', '200'), - ('WORKFLOW_POLLER_BEHAVIUR_MINIMUM', '10'), - ('WORKFLOW_POLLER_BEHAVIUR_INITIAL', '100'), - ('WORKFLOW_POLLER_BEHAVIUR_MAXIMUM', '200'), - ('ACTIVITY_POLLER_BEHAVIUR_MINIMUM', '10'), - ('ACTIVITY_POLLER_BEHAVIUR_INITIAL', '100'), - ('ACTIVITY_POLLER_BEHAVIUR_MAXIMUM', '200'), + ('WORKFLOW_POLLER_BEHAVIOUR_MINIMUM', '10'), + ('WORKFLOW_POLLER_BEHAVIOUR_INITIAL', '100'), + ('WORKFLOW_POLLER_BEHAVIOUR_MAXIMUM', '200'), + ('ACTIVITY_POLLER_BEHAVIOUR_MINIMUM', '10'), + ('ACTIVITY_POLLER_BEHAVIOUR_INITIAL', '100'), + ('ACTIVITY_POLLER_BEHAVIOUR_MAXIMUM', '200'), ] @@ -60,13 +60,13 @@ def prepare_worker( ], max_cached_workflows=local_workflow_parameters['MAX_CACHED_WORKFLOWS'], workflow_task_poller_behavior=PollerBehaviorAutoscaling( - minimum=local_workflow_parameters['WORKFLOW_POLLER_BEHAVIUR_MINIMUM'], - initial=local_workflow_parameters['WORKFLOW_POLLER_BEHAVIUR_INITIAL'], - maximum=local_workflow_parameters['WORKFLOW_POLLER_BEHAVIUR_MAXIMUM'], + minimum=local_workflow_parameters['WORKFLOW_POLLER_BEHAVIOUR_MINIMUM'], + initial=local_workflow_parameters['WORKFLOW_POLLER_BEHAVIOUR_INITIAL'], + maximum=local_workflow_parameters['WORKFLOW_POLLER_BEHAVIOUR_MAXIMUM'], ), activity_task_poller_behavior=PollerBehaviorAutoscaling( - minimum=local_workflow_parameters['ACTIVITY_POLLER_BEHAVIUR_MINIMUM'], - initial=local_workflow_parameters['ACTIVITY_POLLER_BEHAVIUR_INITIAL'], - maximum=local_workflow_parameters['ACTIVITY_POLLER_BEHAVIUR_MAXIMUM'], + minimum=local_workflow_parameters['ACTIVITY_POLLER_BEHAVIOUR_MINIMUM'], + initial=local_workflow_parameters['ACTIVITY_POLLER_BEHAVIOUR_INITIAL'], + maximum=local_workflow_parameters['ACTIVITY_POLLER_BEHAVIOUR_MAXIMUM'], ), ) From 4f5832cfb3d6fd0dcd475c44ffab8c6d987569e8 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Fri, 23 Jan 2026 13:10:16 -0300 Subject: [PATCH 9/9] SIENTIAPDE-1478 Update sientia-dataops-library dependency to version 1.8.2 in requirements.txt --- requirements.txt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/requirements.txt b/requirements.txt index 9225226..11e12b2 100644 --- a/requirements.txt +++ b/requirements.txt @@ -4,5 +4,5 @@ sqlalchemy redis pymongo jinja2 -git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.8.0 +git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.8.2 prometheus-client