From 5dc49b476a68c19eeed17144f50b9ba4ed2e708a Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Tue, 3 Jun 2025 17:29:00 -0300 Subject: [PATCH] SIENTIAPDE-1030 Add comprehensive tests and utility functions for orchestrator activities - Introduced tests for the new Formatters class, covering methods for processing schedules and slots. - Enhanced SlotManager tests with update and delete slot functionalities. - Added TemporalManager tests for creating, updating, and deleting schedules, including frequency parsing. - Implemented utility functions for orchestrator operations, including frequency parsing and tag configuration building. - Created tests for utility functions to ensure correct behavior and integration with orchestrator activities. - Established a new converters module for parsing frequency strings into seconds. --- .env | 4 - docker-compose.yml | 57 ++ orchestrator/activities/activities.py | 7 +- orchestrator/activities/formatters.py | 369 ++++++++++--- orchestrator/activities/slot_manager.py | 106 +++- orchestrator/activities/temporal_manager.py | 205 ++++++- orchestrator/utils/connectors_config.py | 42 +- orchestrator/utils/converters.py | 14 + orchestrator/utils/orchestrator_functions.py | 162 ++++++ orchestrator/worker/worker.py | 84 ++- orchestrator/workflows/orchestrator.py | 125 ++++- test.ipynb | 501 +----------------- .../activities/test_activities.py | 12 +- .../activities/test_formatters.py | 480 +++++++++++++++++ .../activities/test_slot_manager.py | 65 ++- .../activities/test_temporal_manager.py | 210 +++++++- .../utils/test_orchestrator_functions.py | 325 ++++++++++++ .../workflows/test_orchestrator.py | 108 ++++ 18 files changed, 2208 insertions(+), 668 deletions(-) delete mode 100644 .env create mode 100644 orchestrator/utils/converters.py create mode 100644 orchestrator/utils/orchestrator_functions.py create mode 100644 tests/orchestrator/activities/test_formatters.py create mode 100644 tests/orchestrator/utils/test_orchestrator_functions.py diff --git a/.env b/.env deleted file mode 100644 index af9d68d..0000000 --- a/.env +++ /dev/null @@ -1,4 +0,0 @@ -# === Simulator Git Repo === -# Use SSH format because the Dockerfile uses SSH to clone -SIMULATOR_GIT_REPO=git@github.com:Aignosi/sientia-dataops-opc_simulator.git -SIMULATOR_GIT_BRANCH=main diff --git a/docker-compose.yml b/docker-compose.yml index eb96070..baab896 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -15,6 +15,8 @@ services: networks: - sientia-network + + couchbase: image: couchbase/server:7.2.0 container_name: couchbase @@ -62,6 +64,59 @@ services: networks: - sientia-network + kafka: + image: bitnami/kafka:3.7 # Using a specific Kafka version for stability + container_name: kafka + ports: + - "9092:9092" # For clients connecting from the host machine or outside Docker network + environment: + # KRaft (Kafka Raft without Zookeeper) settings + KAFKA_CFG_NODE_ID: '0' + KAFKA_CFG_PROCESS_ROLES: 'broker,controller' + KAFKA_CFG_CONTROLLER_LISTENER_NAMES: 'CONTROLLER' + # Listeners: ://: + # PLAINTEXT_EXTERNAL for host access, INTERNAL for container-to-container communication + KAFKA_CFG_LISTENERS: 'PLAINTEXT_EXTERNAL://0.0.0.0:9092,INTERNAL://0.0.0.0:19092,CONTROLLER://0.0.0.0:9093' + # Advertised Listeners: How clients (including Kafka-UI) will connect + KAFKA_CFG_ADVERTISED_LISTENERS: 'PLAINTEXT_EXTERNAL://localhost:9092,INTERNAL://kafka:19092' + KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: 'CONTROLLER:PLAINTEXT,PLAINTEXT_EXTERNAL:PLAINTEXT,INTERNAL:PLAINTEXT' + KAFKA_CFG_INTER_BROKER_LISTENER_NAME: 'INTERNAL' + KAFKA_CFG_CONTROLLER_QUORUM_VOTERS: '0@kafka:9093' # Node 0 is at kafka:9093 for controller comms + + # Single node cluster settings (important for KRaft single node) + KAFKA_CFG_OFFSETS_TOPIC_REPLICATION_FACTOR: '1' + KAFKA_CFG_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: '1' + KAFKA_CFG_TRANSACTION_STATE_LOG_MIN_ISR: '1' + KAFKA_CFG_DEFAULT_REPLICATION_FACTOR: '1' # For auto-created topics + KAFKA_CFG_NUM_PARTITIONS: '1' # Default partitions for auto-created topics + + KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE: 'true' # Convenient for development + volumes: + - kafka_data:/bitnami/kafka # Bitnami Kafka data directory + networks: + - sientia-network + healthcheck: + # Checks if Kafka is ready by trying to list topics using the internal listener + test: ["CMD-SHELL", "/opt/bitnami/kafka/bin/kafka-topics.sh --bootstrap-server kafka:19092 --list > /dev/null || exit 1"] + interval: 30s + timeout: 10s + retries: 5 + + kafka-ui: + image: provectuslabs/kafka-ui:latest + container_name: kafka-ui + ports: + - "8082:8080" # Kafka UI will be accessible on host's port 8082 + environment: + KAFKA_CLUSTERS_0_NAME: sientia-local-kafka + KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:19092 # Connects to Kafka's internal listener + # DYNAMIC_CONFIG_ENABLED: 'true' # Optional: To allow config changes through UI + depends_on: + kafka: # Ensures Kafka starts and is healthy before Kafka UI + condition: service_healthy + networks: + - sientia-network + networks: sientia-network: @@ -73,4 +128,6 @@ volumes: couchbase_data: driver: local redis_data: + driver: local + kafka_data: driver: local \ No newline at end of file diff --git a/orchestrator/activities/activities.py b/orchestrator/activities/activities.py index 9579ec3..1265539 100644 --- a/orchestrator/activities/activities.py +++ b/orchestrator/activities/activities.py @@ -5,12 +5,13 @@ with workflow.unsafe.imports_passed_through(): from orchestrator.activities.couchbase import Couchbase from orchestrator.activities.temporal_manager import TemporalManager from orchestrator.activities.slot_manager import SlotManager + from orchestrator.activities.formatters import Formatters from typing import Any from logging import Logger from sientia_do.notifications.handlers import NotificationHandler -class Activities(Couchbase, TemporalManager, SlotManager): +class Activities(Couchbase, TemporalManager, SlotManager, Formatters): def __init__(self, temporal_client: Client, @@ -39,6 +40,10 @@ class Activities(Couchbase, TemporalManager, SlotManager): logger=logger, notification_handler=notification_handler) + Formatters.__init__(self, + logger=logger, + notification_handler=notification_handler) + @activity.defn(name="prepare_activity") async def prepare_activity(self, input_data: dict[str, Any]): await super().prepare_activity(input_data) diff --git a/orchestrator/activities/formatters.py b/orchestrator/activities/formatters.py index 9d4ceb1..d88bd5b 100644 --- a/orchestrator/activities/formatters.py +++ b/orchestrator/activities/formatters.py @@ -1,10 +1,16 @@ +from sientia_do.notifications.models import NotificationLevel from temporalio import activity, workflow with workflow.unsafe.imports_passed_through(): - from sientia_do.temporal.activities.base import BaseActivity + import json from typing import Any from logging import Logger from sientia_do.notifications.handlers import NotificationHandler + from sientia_do.temporal.activities.base import BaseActivity + from orchestrator.utils.orchestrator_functions import ( + scouter, predictions_batch, gather_read_tags, build_tag_config + ) + from math import ceil class Formatters(BaseActivity): @@ -13,7 +19,22 @@ class Formatters(BaseActivity): notification_handler=notification_handler) @activity.defn(name="process_schedules") - async def process_schedules(self, input_data: dict[str, Any]): + async def process_schedules(self, input_data: dict[str, Any]) -> dict[str, Any]: + """ + Process schedules. Generates a schedule config dictionary + based on the input data workflow type. + + Args: + - input_data (dict[str, Any]): The input data containing + the schedules to process. + - pipelines (list[dict[str, Any]]): The schedules to process. + + Returns: + - dict[str, Any]: The schedule config dictionary + """ + + self.logger.info("Processing schedules...") + pipelines = input_data['pipelines'] schedule_config = {} @@ -21,115 +42,291 @@ class Formatters(BaseActivity): for pipeline in pipelines: if pipeline['workflow_type'] == 'scouter': schedule_config[pipeline['schedule_name']] = scouter(pipeline) + elif pipeline['workflow_type'] == 'predictions_batch': + schedule_config[pipeline['schedule_name'] + ] = predictions_batch(pipeline) return schedule_config + @activity.defn(name="process_slots") + async def process_slots(self, input_data: dict[str, Any]) -> dict[str, Any]: + """ + Extracts all read tags from input pipelines, divides them into slots and + returns a slot config dictionary. If no ingestor is available, only one slot + is created. -def common_config(config: dict[str, Any]): - return { - "workflow_type": "scouter", - "schedule_name": config['schedule_name'], - "frequency": config.get('frequency', '1m'), - "max_retry_policy": config.get('max_retry_policy', 1), + Args: + - input_data (dict[str, Any]): The input data containing + the schedules to process. + - pipelines (list[dict[str, Any]]): The schedules to process. + - opc_servers (list[str]): The OPC servers to create ingestor config. + - active_ingestors (list[str]): The active ingestors to divide into slots. - "model_id": config['model_id'], - "model_name": config['model_name'], - } + Returns: + - dict[str, Any]: The slot config dictionary + """ + self.logger.info("Processing slots...") -def scouter(config: dict[str, Any]): - filters = {} - for f in config['filters']: - filters[f['filter_name']] = { - "policy": f['policy'] + pipelines = input_data['pipelines'] + opc_servers_list = input_data['opc_servers'] + active_ingestors = input_data['active_ingestors'] + + opc_servers = {} + for server in opc_servers_list: + opc_servers[server['server_name']] = { + **server, + } + + tags = list(gather_read_tags(pipelines).values()) + + number_of_tags = len(tags) + number_of_slots = len(active_ingestors) if active_ingestors else 1 + tags_per_slot = ceil(number_of_tags / number_of_slots) + + slot_config = {} + last_index = 0 + + for i in range(1, number_of_slots): + slot_config[f"{i}"] = {} + for tag in tags[last_index:last_index + tags_per_slot]: + slot_config = build_tag_config( + tag, slot_config.copy(), opc_servers, i) + last_index += tags_per_slot + + slot_config[f"{number_of_slots}"] = {} + for tag in tags[last_index:]: + slot_config = build_tag_config( + tag, slot_config.copy(), opc_servers, number_of_slots) + + return slot_config + + @activity.defn(name="create_schedule_config") + async def create_schedule_config(self, + input_data: dict[str, Any]) -> dict[str, Any]: + """ + Creates a schedule config dictionary based on the input data. + Checks the existing schedule config and updates it with the new schedule config, + deleting unnecessary schedules, creating new schedules and updating existing schedules. + + Args: + - input_data (dict[str, Any]): The input data containing + the schedules to process. + - current_schedule_config (dict[str, Any]): The current schedule + config in Temporal server. + - schedule_config (dict[str, Any]): The schedule config to process. + + Returns: + - dict[str, Any]: The schedule config dictionary + """ + + self.logger.info("Creating schedule config...") + + current_schedule_config = input_data['current_schedule_config'] + schedule_config = input_data['schedule_config'] + + to_update = {} + to_create = {} + to_delete = [] + + for schedule_name, schedule in schedule_config.items(): + if schedule_name in current_schedule_config: + if schedule != current_schedule_config[schedule_name]['data']: + to_update[schedule_name] = schedule + + elif schedule_name not in current_schedule_config: + to_create[schedule_name] = schedule + + for schedule_name in current_schedule_config: + if schedule_name not in schedule_config: + to_delete.append(schedule_name) + + return { + "to_update": to_update, + "to_create": to_create, + "to_delete": to_delete } - tags = {} - for tag in config['read_tags']: - tags[tag['tag_name']] = { - "aggr_func": tag.get('aggr_func', 'lts'), - "data_range": tag.get('data_range', [-100, 100]) + @activity.defn(name="create_slot_config") + async def create_slot_config(self, + input_data: dict[str, Any]) -> dict[str, Any]: + """ + Creates a slot config dictionary based on the input data. + Checks the existing slot config and updates it with the new slot config, + deleting unnecessary slots. + + Args: + - input_data (dict[str, Any]): The input data containing + the slots to process. + - current_slot_config (dict[str, Any]): The current slot + config in Temporal server. + - slot_config (dict[str, Any]): The slot config to process. + + Returns: + - dict[str, Any]: The slot config dictionary + """ + + self.logger.info("Creating slot config...") + + current_slot_config = input_data['current_slot_config'] + slot_config = input_data['slot_config'] + to_delete = [] + + number_of_current_slots = len(current_slot_config) + number_of_slots = len(slot_config) + + if number_of_current_slots > number_of_slots: + to_delete = [str(i) for i in range( + number_of_slots + 1, number_of_current_slots + 1)] + + return { + "to_delete": to_delete, + "to_insert": slot_config } - return { - **common_config(config), + def send_success_report(self, message: str, notification_id: str) -> None: + self.notification_handler.build_and_send_notification( + notification_id, + message, + "report_orchestration", + NotificationLevel.INFO + ) - "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 - } + def send_error_report(self, message: str, notification_id: str, + attachment: dict[str, Any]) -> None: + self.notification_handler.build_and_send_notification( + notification_id, + message, + "report_orchestration", + NotificationLevel.ERROR, + attachment_content=json.dumps(attachment, indent=4, sort_keys=True) + ) + def parse_report(self, input_data: dict[str, Any]) -> tuple[list[str], list[str]]: + success_keys = [key for key, value + in input_data.items() if value['success']] -def overlap_filter_config(base_filter_config: dict[str, Any], config: dict[str, Any]): - for fil in config['filters']: - base_filter_config[fil['filter_name']] = { - "policy": fil['policy'], - "config": fil.get('config', {}) - } + error_keys = [key for key, value + in input_data.items() if not value['success']] - return base_filter_config + return success_keys, error_keys + @activity.defn(name="report_schedule_orchestration") + async def report_schedule_orchestration(self, + input_data: dict[str, Any]) -> None: + """ + Reports the orchestration result to the notification handler. -def process_path_priority(path_priority: list[str]): - for priority in path_priority[:]: - if priority not in ["STOP", "CONTINUE", "REPEAT"]: - path_priority.remove(priority) + Args: + - input_data (dict[str, Any]): The input data containing + the orchestration result. + - created_schedules (dict[str, Any]): The created schedules. + - updated_schedules (dict[str, Any]): The updated schedules. + - deleted_schedules (list[str]): The deleted schedules. + """ - if len(path_priority) != 3: - for priority in ["STOP", "CONTINUE", "REPEAT"]: - if priority not in path_priority: - path_priority.append(priority) + self.logger.info("Reporting orchestration...") - return path_priority + created_schedules = input_data['created_schedules'] + updated_schedules = input_data['updated_schedules'] + deleted_schedules = input_data['deleted_schedules'] + # Send report for created schedules + if len(created_schedules) > 0: + success_keys, error_keys = self.parse_report(created_schedules) -def predictions_batch(config: dict[str, Any]): - tags = {} - for tag in config['write_tags']: - if tag['server_name'] not in tags: - tags[tag['server_name']] = {} + if len(success_keys) > 0: + self.send_success_report( + f"Created schedules: \n {', '.join(success_keys)}", + "REPORT_ORCHESTRATION_CREATED_SCHEDULES" + ) - tag_type = tag['type'] + if len(error_keys) > 0: + self.send_error_report( + f"Failed to create schedules: \n {', '.join(error_keys)}", + "REPORT_ORCHESTRATION_CREATED_SCHEDULES", + created_schedules + ) - if tag_type == 'prediction' or tag_type == 'confidence': - tag_type_str = f"{tag_type}_tags" + # Send report for updated schedules + if len(updated_schedules) > 0: + success_keys, error_keys = self.parse_report(updated_schedules) - if tag_type_str not in tags[tag['server_name']]: - tags[tag['server_name']][tag_type_str] = {} + if len(success_keys) > 0: + self.send_success_report( + f"Updated schedules: \n {', '.join(success_keys)}", + "REPORT_ORCHESTRATION_UPDATED_SCHEDULES" + ) - tags[tag['server_name']][tag_type_str][tag['addr']] = { - "data_type": tag.get('data_type', 'float'), - } + if len(error_keys) > 0: + self.send_error_report( + f"Failed to update schedules: \n {', '.join(error_keys)}", + "REPORT_ORCHESTRATION_UPDATED_SCHEDULES", + updated_schedules + ) - path_priority = process_path_priority(config.get( - 'path_priority', ["STOP", "CONTINUE", "REPEAT"])) + if len(deleted_schedules) > 0: + success_keys, error_keys = self.parse_report(deleted_schedules) - return { - **common_config(config), + if len(success_keys) > 0: + self.send_success_report( + f"Deleted schedules: \n {', '.join(success_keys)}", + "REPORT_ORCHESTRATION_DELETED_SCHEDULES" + ) - "query": config['query'], - "schema": "sientia_data", - "table_name": "predictions", - "retention_time": config.get('model_retention_minutes', 60) * 60, - "opc_output_config": tags, - "input_filters": overlap_filter_config({ - "EMPTY_DATA": { - "policy": "STOP" - } - }, config['input_filters']), - "mlflow_transform_filters": overlap_filter_config({ - "API_ERROR": { - "policy": "STOP" - } - }, config['mlflow_transform_filters']), - "mlflow_predict_filters": overlap_filter_config({ - "API_ERROR": { - "policy": "STOP" - } - }, config['mlflow_predict_filters']), - "path_priority": path_priority - } + if len(error_keys) > 0: + self.send_error_report( + f"Failed to delete schedules: \n {', '.join(error_keys)}", + "REPORT_ORCHESTRATION_DELETED_SCHEDULES", + deleted_schedules + ) + + @activity.defn(name="report_slot_orchestration") + async def report_slot_orchestration(self, + input_data: dict[str, Any]) -> None: + """ + Reports the orchestration result to the notification handler. + + Args: + - input_data (dict[str, Any]): The input data containing + the orchestration result. + - inserted_slots (dict[str, Any]): The inserted slots. + - deleted_slots (list[str]): The deleted slots. + """ + + self.logger.info("Reporting orchestration...") + + inserted_slots = input_data['inserted_slots'] + deleted_slots = input_data['deleted_slots'] + + if len(inserted_slots) > 0: + success_keys, error_keys = self.parse_report(inserted_slots) + + if len(success_keys) > 0: + self.send_success_report( + f"Inserted slots: \n {', '.join(success_keys)}", + "REPORT_ORCHESTRATION_INSERTED_SLOTS" + ) + + if len(error_keys) > 0: + self.send_error_report( + f"Failed to insert slots: \n {', '.join(error_keys)}", + "REPORT_ORCHESTRATION_INSERTED_SLOTS", + inserted_slots + ) + + if len(deleted_slots) > 0: + success_keys, error_keys = self.parse_report(deleted_slots) + + if len(success_keys) > 0: + self.send_success_report( + f"Deleted slots: \n {', '.join(success_keys)}", + "REPORT_ORCHESTRATION_DELETED_SLOTS" + ) + + if len(error_keys) > 0: + self.send_error_report( + f"Failed to delete slots: \n {', '.join(error_keys)}", + "REPORT_ORCHESTRATION_DELETED_SLOTS", + deleted_slots + ) diff --git a/orchestrator/activities/slot_manager.py b/orchestrator/activities/slot_manager.py index f6ab941..45d77d4 100644 --- a/orchestrator/activities/slot_manager.py +++ b/orchestrator/activities/slot_manager.py @@ -32,23 +32,20 @@ class SlotManager(Redis): slot_keys = self.redis_client.keys("slot:opc_tags:*") - if slot_keys: - decoded_keys = [key.decode('utf-8') for key in slot_keys] - values = self.redis_client.mget(decoded_keys) + self.logger.debug("Slot keys: %s", slot_keys) - for i, key in enumerate(decoded_keys): - value = values[i] - if value is not None: - try: - opc_slots[key] = value.decode('utf-8') - except (UnicodeDecodeError, AttributeError): - opc_slots[key] = value - else: - opc_slots[key] = None + if slot_keys: + if isinstance(slot_keys[0], bytes): + decoded_keys = [key.decode('utf-8') for key in slot_keys] + else: + decoded_keys = slot_keys + + for key in decoded_keys: + opc_slots[key] = self.get(key) self.logger.info(f"Loaded {len(opc_slots)} OPC slots") - self.logger.debug("OPC slots: \n %s", + self.logger.debug("Loaded: \n %s", json.dumps(opc_slots, indent=4, sort_keys=True)) return opc_slots @@ -71,3 +68,86 @@ class SlotManager(Redis): self.logger.debug("Active ingestors: \n %s", active_ingestors) return [ingestor.decode('utf-8') for ingestor in active_ingestors] + + @activity.defn(name="update_slots") + async def update_slots(self, input_data: dict[str, Any]) -> dict[str, Any]: + """ + Update OPC slots in Redis + + Args: + - input_data (dict[str, Any]): The input data containing + the slots to update. + - to_insert (dict[str, Any]): The slots to insert. + + Returns: + - report (dict[str, Any]): A report of the updated slots. + """ + + to_insert = input_data['to_insert'] + self.logger.info("Updating OPC slots...") + + report = {} + + for slot in to_insert: + try: + self.set(f"slot:opc_tags:{slot}", + to_insert[slot], ttl=None) + report[slot] = { + "success": True, + "message": "Slot updated successfully" + } + except Exception as e: + self.logger.error("Failed to update slot %s: %s", + slot, str(e)) + report[slot] = { + "success": False, + "message": str(e) + } + + self.logger.info(f"Updated {len(to_insert)} OPC slots") + + self.logger.debug("Report: \n %s", + json.dumps(report, indent=4, sort_keys=True)) + + return report + + @activity.defn(name="delete_slots") + async def delete_slots(self, input_data: dict[str, Any]) -> dict[str, Any]: + """ + Delete OPC slots from Redis + + Args: + - input_data (dict[str, Any]): The input data containing + the slots to delete. + - to_delete (list[str]): The slots to delete. + + Returns: + - report (dict[str, Any]): A report of the deleted slots. + """ + + to_delete = input_data['to_delete'] + self.logger.info("Deleting OPC slots...") + + report = {} + + for slot in to_delete: + try: + self.redis_client.delete(f"slot:opc_tags:{slot}") + report[slot] = { + "success": True, + "message": "Slot deleted successfully" + } + except Exception as e: + self.logger.error("Failed to delete slot %s: %s", + slot, str(e)) + report[slot] = { + "success": False, + "message": str(e) + } + + self.logger.info(f"Deleted {len(to_delete)} OPC slots") + + self.logger.debug("Report: \n %s", + json.dumps(report, indent=4, sort_keys=True)) + + return report diff --git a/orchestrator/activities/temporal_manager.py b/orchestrator/activities/temporal_manager.py index daa8fc5..71fc641 100644 --- a/orchestrator/activities/temporal_manager.py +++ b/orchestrator/activities/temporal_manager.py @@ -1,5 +1,7 @@ from temporalio import activity, workflow -from temporalio.client import Client +from temporalio.client import ( + Client, Schedule, ScheduleActionStartWorkflow, ScheduleIntervalSpec, ScheduleSpec, ScheduleUpdate, ScheduleUpdateInput) +from temporalio.common import SearchAttributeKey, SearchAttributePair, TypedSearchAttributes with workflow.unsafe.imports_passed_through(): from typing import Any @@ -8,7 +10,9 @@ with workflow.unsafe.imports_passed_through(): from sientia_do.temporal.activities.base import BaseActivity from google.protobuf.json_format import MessageToDict import base64 + from datetime import timedelta import json + from orchestrator.utils.converters import parse_frequency class TemporalManager(BaseActivity): @@ -17,6 +21,13 @@ class TemporalManager(BaseActivity): self.temporal_client = temporal_client + self.schedule_handles = {} + + self.model_id_id_key = SearchAttributeKey.for_keyword("model_id") + self.model_name_id_key = SearchAttributeKey.for_keyword("model_name") + self.orchestrated_id_key = SearchAttributeKey.for_keyword( + "orchestrated") + BaseActivity.__init__(self, logger=logger, notification_handler=notification_handler) @@ -41,7 +52,9 @@ class TemporalManager(BaseActivity): if search_attrs.get("Orchestrated", ["false"]) == ["true"]: schedule_id = schedule.id - handle = self.temporal_client.get_schedule(schedule_id) + handle = self.temporal_client.get_schedule_handle(schedule_id) + + self.schedule_handles[schedule_id] = handle desc = await handle.describe() @@ -54,7 +67,6 @@ class TemporalManager(BaseActivity): orchestrated_schedules[schedule_id] = { 'frequency': frequency, 'data': json.loads(data), - 'handle': handle } self.logger.info("Found %d orchestrated schedules", @@ -64,3 +76,190 @@ class TemporalManager(BaseActivity): orchestrated_schedules) return orchestrated_schedules + + @activity.defn(name="create_schedules") + async def create_schedules(self, input_data: dict[str, Any]) -> dict[str, Any]: + """ + Create schedules in Temporal + + Args: + - input_data (dict[str, Any]): The input data containing + the schedules to create. + - schedules (dict[str, Any]): The schedules to create. + + Returns: + - dict[str, Any]: A report of the created schedules. + """ + + schedules = input_data['schedules'] + + report = {} + + for schedule_name, schedule in schedules.items(): + search_attributes = TypedSearchAttributes([ + SearchAttributePair( + key=self.model_id_id_key, + value=schedule['model_id'] + ), + SearchAttributePair( + key=self.model_name_id_key, + value=schedule['model_name'] + ), + SearchAttributePair( + key=self.orchestrated_id_key, + value="true" + ) + ]) + workflow_type = schedule['workflow_type'] + + try: + await self.temporal_client.create_schedule( + schedule_name, + Schedule( + action=ScheduleActionStartWorkflow( + workflow=workflow_type, + args=schedule, + id=schedule_name, + task_queue=f"{workflow_type}-queue" + ), + spec=ScheduleSpec( + intervals=[ + ScheduleIntervalSpec( + every=timedelta(seconds=parse_frequency( + schedule.get('frequency', '1m'))) + ) + ] + ) + ), + search_attributes=search_attributes + ) + + report[schedule_name] = { + "success": True, + "message": "Schedule created successfully" + } + except Exception as e: + self.logger.error("Failed to create schedule %s: %s", + schedule_name, str(e)) + report[schedule_name] = { + "success": False, + "message": str(e) + } + + self.logger.info(f"Processed {len(schedules)} schedules") + + self.logger.debug("\n %s", + json.dumps(report, indent=4, sort_keys=True)) + + return report + + @activity.defn(name="update_schedules") + async def update_schedules(self, input_data: dict[str, Any]) -> dict[str, Any]: + """ + Update schedules in Temporal + + Args: + - input_data (dict[str, Any]): The input data containing + the schedules to update. + - schedules (dict[str, Any]): The schedules to update. + + Returns: + - dict[str, Any]: A report of the updated schedules. + """ + + schedules = input_data['schedules'] + + report = {} + + for schedule_name, schedule in schedules.items(): + try: + handler = self.schedule_handles.get(schedule_name) + + if not handler: + raise ValueError(f"Schedule {schedule_name} not found") + + async def update_schedule(input_data: ScheduleUpdateInput) -> ScheduleUpdate: + schedule_action = input_data.description.schedule.action + + if hasattr(schedule_action, "args"): + schedule_action.args = schedule + + input_data.description.schedule.spec.intervals = [ + ScheduleIntervalSpec( + every=timedelta(seconds=parse_frequency( + schedule.get('frequency', '1m'))) + ) + ] + + return ScheduleUpdate(schedule=input_data.description.schedule) + + await handler.update(update_schedule) + + del update_schedule + + report[schedule_name] = { + "success": True, + "message": "Schedule updated successfully" + } + except Exception as e: + self.logger.error("Failed to update schedule %s: %s", + schedule_name, str(e)) + report[schedule_name] = { + "success": False, + "message": str(e) + } + + self.logger.info(f"Processed {len(schedules)} schedules") + + self.logger.debug("\n %s", + json.dumps(report, indent=4, sort_keys=True)) + + return report + + @activity.defn(name="delete_schedules") + async def delete_schedules(self, input_data: dict[str, Any]) -> dict[str, Any]: + """ + Delete schedules in Temporal + + Args: + - input_data (dict[str, Any]): The input data containing + the schedules to delete. + - schedules (list[str]): The schedules to delete. + + Returns: + - dict[str, Any]: A report of the deleted schedules. + """ + + schedules = input_data['schedules'] + + report = {} + + for schedule_name in schedules: + try: + handler = self.schedule_handles.get(schedule_name) + + if not handler: + raise ValueError(f"Schedule {schedule_name} not found") + + await handler.delete() + + del self.schedule_handles[schedule_name] + + report[schedule_name] = { + "success": True, + "message": "Schedule deleted successfully" + } + except Exception as e: + self.logger.error("Failed to delete schedule %s: %s", + schedule_name, str(e)) + report[schedule_name] = { + "success": False, + "message": str(e) + } + + self.logger.info(f"Processed {len(schedules)} schedules") + + self.logger.debug("\n %s", + json.dumps(report, indent=4, sort_keys=True)) + + return report diff --git a/orchestrator/utils/connectors_config.py b/orchestrator/utils/connectors_config.py index 80ed24d..2e5e80f 100644 --- a/orchestrator/utils/connectors_config.py +++ b/orchestrator/utils/connectors_config.py @@ -1,42 +1,18 @@ from os import getenv -import json -def build_postgres_config(): +def build_redis_config(): return { - 'host': getenv('POSTGRES_HOST', 'localhost'), - 'port': int(getenv('POSTGRES_PORT', '5432')), - 'user': getenv('POSTGRES_USER', 'sientia'), - 'password': getenv('POSTGRES_PASSWORD', 'sientia'), - 'dbname': getenv('POSTGRES_DBNAME', 'sientia'), - 'min_connections': int(getenv('POSTGRES_MIN_CONNECTIONS', '5')), - 'max_connections': int(getenv('POSTGRES_MAX_CONNECTIONS', '20')) + 'host': getenv('REDIS_HOST', 'localhost'), + 'port': int(getenv('REDIS_PORT', '6379')), + 'username': getenv('REDIS_USERNAME', None), + 'password': getenv('REDIS_PASSWORD', None) } -def build_mlflow_config(): +def build_couchbase_config(): return { - 'host': getenv('MLFLOW_HOST', 'http://localhost'), - 'port': int(getenv('MLFLOW_PORT', '5080')), - 'username': getenv('MLFLOW_USERNAME', 'aignosi'), - 'password': getenv('MLFLOW_PASSWORD', 'aignosi') - } - - -def build_opc_config(): - opc_raw = getenv('OPC_CONFIG', None) - - if opc_raw: - return json.loads(opc_raw) - - return { - 'opc': { - 'name': getenv('OPC_NAME', 'opc'), - 'url': getenv('OPC_URL', 'opc.tcp://localhost:4840'), - 'server_uri': getenv('OPC_SERVER_URI', 'opc.tcp://localhost:4840'), - 'cert_path': getenv('OPC_CERT_PATH', None), - 'private_key_path': getenv('OPC_PRIVATE_KEY_PATH', None), - 'server_cert_path': getenv('OPC_SERVER_CERT_PATH', None), - 'reconnection_interval': int(getenv('OPC_RECONNECTION_INTERVAL', '120')) - } + 'connection_string': getenv('COUCHBASE_CONNECTION_STRING', 'couchbase://localhost'), + 'username': getenv('COUCHBASE_USERNAME', 'sientia'), + 'password': getenv('COUCHBASE_PASSWORD', 'sientia') } diff --git a/orchestrator/utils/converters.py b/orchestrator/utils/converters.py new file mode 100644 index 0000000..45bd436 --- /dev/null +++ b/orchestrator/utils/converters.py @@ -0,0 +1,14 @@ +def parse_frequency(frequency: str) -> int: + """ + Parse frequency string to seconds + """ + if frequency.endswith("s"): + return int(frequency[:-1]) + elif frequency.endswith("m"): + return int(frequency[:-1]) * 60 + elif frequency.endswith("h"): + return int(frequency[:-1]) * 60 * 60 + elif frequency.endswith("d"): + return int(frequency[:-1]) * 60 * 60 * 24 + else: + raise ValueError("Invalid frequency") diff --git a/orchestrator/utils/orchestrator_functions.py b/orchestrator/utils/orchestrator_functions.py new file mode 100644 index 0000000..5e57968 --- /dev/null +++ b/orchestrator/utils/orchestrator_functions.py @@ -0,0 +1,162 @@ +from typing import Any + + +def common_config(config: dict[str, Any]): + return { + "workflow_type": config['workflow_type'], + "schedule_name": config['schedule_name'], + "frequency": config.get('frequency', '1m'), + "max_retry_policy": config.get('max_retry_policy', 1), + + "model_id": config['model_id'], + "model_name": config['model_name'], + } + + +def scouter(config: dict[str, Any]): + filters = {} + for f in config.get('filters', []): + filters[f['filter_name']] = { + "policy": f['policy'] + } + + tags = {} + for tag in config['read_tags']: + tags[tag['tag_name']] = { + "aggr_func": tag.get('aggr_func', 'lts'), + "data_range": tag.get('data_range', [-100, 100]) + } + + return { + **common_config(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 + } + + +def overlap_filter_config(base_filter_config: dict[str, Any], config: list[dict[str, Any]]): + for fil in config: + base_filter_config[fil['filter_name']] = { + "policy": fil['policy'], + "config": fil.get('config', {}) + } + + return base_filter_config + + +def process_path_priority(path_priority: list[str]): + for priority in path_priority[:]: + if priority not in ["STOP", "CONTINUE", "REPEAT"]: + path_priority.remove(priority) + + for priority in ["STOP", "CONTINUE", "REPEAT"]: + if priority not in path_priority: + path_priority.append(priority) + + return path_priority[0:3] + + +def predictions_batch(config: dict[str, Any]): + tags = {} + for tag in config['write_tags']: + if tag['server_name'] not in tags: + tags[tag['server_name']] = {} + + tag_type = tag['type'] + + if tag_type == 'prediction' or tag_type == 'confidence': + tag_type_str = f"{tag_type}_tags" + + if tag_type_str not in tags[tag['server_name']]: + tags[tag['server_name']][tag_type_str] = {} + + tags[tag['server_name']][tag_type_str][tag['addr']] = { + "data_type": tag.get('data_type', 'float'), + } + + path_priority = process_path_priority(config.get( + 'path_priority', ["STOP", "CONTINUE", "REPEAT"])) + + return { + **common_config(config), + + "query": config['query'], + "schema": "sientia_data", + "table_name": "predictions", + "retention_time": config.get('model_retention_minutes', 60) * 60, + "opc_output_config": tags, + "input_filters": overlap_filter_config({ + "EMPTY_DATA": { + "policy": "STOP", + "config": {} + } + }, config['input_filters']), + "mlflow_transform_filters": overlap_filter_config({ + "API_ERROR": { + "policy": "STOP", + "config": {} + } + }, config['mlflow_transform_filters']), + "mlflow_predict_filters": overlap_filter_config({ + "API_ERROR": { + "policy": "STOP", + "config": {} + } + }, config['mlflow_predict_filters']), + "path_priority": path_priority + } + + +def gather_read_tags(pipelines: list[dict[str, Any]]) -> dict[str, Any]: + """ + Gathers all read tags from input pipelines. + + Args: + - pipelines (list[dict[str, Any]]): The schedules to process. + + Returns: + - dict[str, Any]: The read tags dictionary + """ + + tags = {} + + # Get all read tags from pipelines + for pipeline in pipelines: + for tag in pipeline['read_tags']: + tag_string = f"{tag['server_name']}:{tag['tag_address']}" + if tag_string not in tags: + tags[tag_string] = { + **tag, + "topics": [] + } + + tags[tag_string]['topics'].append( + f"raw_{pipeline['schedule_name']}") + + return tags + + +def build_tag_config(tag: dict[str, Any], slot_config: dict[str, Any], + opc_servers: dict[str, Any], i: int): + server_name = tag['server_name'] + if server_name not in slot_config[f"{i}"]: + slot_config[f"{i}"][server_name] = { + "name": server_name, + "url": opc_servers[server_name]['url'], + "server_uri": opc_servers[server_name]['uri'], + "tags": {} + } + for name, spec in opc_servers[server_name].get('security_spec', {}).items(): + slot_config[f"{i}"][server_name][name] = spec + + slot_config[f"{i}"][server_name]["tags"][tag['tag_address']] = { + **tag, + } + + return slot_config diff --git a/orchestrator/worker/worker.py b/orchestrator/worker/worker.py index 5b902a9..0814b25 100644 --- a/orchestrator/worker/worker.py +++ b/orchestrator/worker/worker.py @@ -1,22 +1,18 @@ from temporalio import workflow, client from temporalio.worker import Worker -import sys with workflow.unsafe.imports_passed_through(): import os + import sys import asyncio - from laborious.workflows.predictions_batch import PredictionsBatch - from laborious.workflows.sub_workflows.prediction_process import PredictionProcess - from laborious.workflows.sub_workflows.format_and_export_prediction import \ - FormatAndExportPrediction - from laborious.activities.activities import Activities - from laborious.utils.logger import get_logger - from laborious.utils.connectors_config import ( - build_postgres_config, - build_mlflow_config, - build_opc_config + from orchestrator.workflows.orchestrator import Orchestrator + from orchestrator.activities.activities import Activities + from orchestrator.utils.connectors_config import ( + build_couchbase_config, + build_redis_config, ) from sientia_do.notifications.handlers import NotificationHandler + from sientia_do.temporal.utils.logger import get_logger async def main(): @@ -30,21 +26,7 @@ async def main(): notification_handler = NotificationHandler( servers=os.getenv('KAFKA_BOOTSTRAP_SERVERS', 'http://localhost:9092'), logger=logger, - project_name=os.getenv('PROJECT_NAME', 'laborious'), - pipeline_name='-', - trigger_name='-', - model_name='-', - model='-' - ) - - logger.info('Starting Activities...') - - activities = Activities( - postgres_config=build_postgres_config(), - mlflow_config=build_mlflow_config(), - opc_config=build_opc_config(), - logger=logger, - notification_handler=notification_handler + project_name=os.getenv('PROJECT_NAME', 'orchestrator'), ) logger.info('Starting Temporal Client...') @@ -54,33 +36,45 @@ async def main(): namespace=os.getenv('TEMPORAL_NAMESPACE', 'default') ) + logger.info('Starting Activities...') + + activities = Activities( + temporal_client=temporal_client, + couchbase_config=build_couchbase_config(), + redis_config=build_redis_config(), + logger=logger, + notification_handler=notification_handler + ) + logger.info('Starting Workers...') workers = [ Worker( temporal_client, - task_queue='predictions-queue', - workflows=[PredictionsBatch, PredictionProcess, - FormatAndExportPrediction], + task_queue='orchestrator-queue', + workflows=[Orchestrator], activities=[ # Base activities.prepare_activity, - # MLFlow - activities.request_predict, - activities.request_transform, - # Gates - activities.input_gate, - activities.mlflow_response_gate, - activities.mlflow_content_gate, - activities.format_prediction, - activities.format_default_prediction, - activities.get_last_timestamp, - # OPC - activities.write_opc_data, - # Postgres - activities.load_custom_query, - activities.repeat_last_prediction, - activities.export_data_to_postgres + # Redis + activities.load_active_ingestors, + activities.load_opc_slots, + activities.update_slots, + activities.delete_slots, + # Couchbase + activities.load_query_from_couchbase, + # Temporal + activities.load_schedule, + activities.create_schedules, + activities.update_schedules, + activities.delete_schedules, + # Formatters + activities.process_schedules, + activities.process_slots, + activities.create_schedule_config, + activities.create_slot_config, + activities.report_schedule_orchestration, + activities.report_slot_orchestration, ] ) ] diff --git a/orchestrator/workflows/orchestrator.py b/orchestrator/workflows/orchestrator.py index ca64d87..12550d5 100644 --- a/orchestrator/workflows/orchestrator.py +++ b/orchestrator/workflows/orchestrator.py @@ -50,7 +50,7 @@ class Orchestrator: start_to_close_timeout=timedelta(seconds=60) ) - slot_config_handler = workflow.execute_local_activity_method( + current_slot_config_handler = workflow.execute_local_activity_method( Activities.load_opc_slots, retry_policy=retry_policy, start_to_close_timeout=timedelta(seconds=60) @@ -64,6 +64,127 @@ class Orchestrator: pipeline_config = await pipeline_config_handler orchestrated_schedules = await orchestrated_schedules_handler - slot_config = await slot_config_handler + current_slot_config = await current_slot_config_handler opc_servers = await opc_servers_handler active_ingestors = await active_ingestors_handler + + schedules_config_handler = workflow.execute_local_activity_method( + Activities.process_schedules, + { + 'pipelines': pipeline_config + }, + retry_policy=retry_policy, + start_to_close_timeout=timedelta(seconds=60) + ) + + slot_config_handler = workflow.execute_local_activity_method( + Activities.process_slots, + { + 'opc_servers': opc_servers, + 'active_ingestors': active_ingestors, + 'pipelines': pipeline_config, + }, + retry_policy=retry_policy, + start_to_close_timeout=timedelta(seconds=60) + ) + + schedules_config = await schedules_config_handler + slot_config = await slot_config_handler + + schedule_actions_handler = workflow.execute_local_activity_method( + Activities.create_schedule_config, + { + 'current_schedule_config': orchestrated_schedules, + 'schedule_config': schedules_config + }, + retry_policy=retry_policy, + start_to_close_timeout=timedelta(seconds=60) + ) + + slot_actions_handler = workflow.execute_local_activity_method( + Activities.create_slot_config, + { + 'current_slot_config': current_slot_config, + 'slot_config': slot_config + }, + retry_policy=retry_policy, + start_to_close_timeout=timedelta(seconds=60) + ) + + schedule_actions = await schedule_actions_handler + slot_actions = await slot_actions_handler + + slot_deletion_report_handler = workflow.execute_activity_method( + Activities.delete_slots, + { + 'to_delete': slot_actions['to_delete'] + }, + retry_policy=retry_policy, + start_to_close_timeout=timedelta(seconds=60) + ) + + slot_insertion_report_handler = workflow.execute_activity_method( + Activities.update_slots, + { + 'to_insert': slot_actions['to_insert'] + }, + retry_policy=retry_policy, + start_to_close_timeout=timedelta(seconds=60) + ) + + schedule_deletion_report_handler = workflow.execute_activity_method( + Activities.delete_schedules, + { + 'schedules': schedule_actions['to_delete'] + }, + retry_policy=retry_policy, + start_to_close_timeout=timedelta(seconds=60) + ) + + schedule_insertion_report_handler = workflow.execute_activity_method( + Activities.create_schedules, + { + 'schedules': schedule_actions['to_create'] + }, + retry_policy=retry_policy, + start_to_close_timeout=timedelta(seconds=60) + ) + + schedule_update_report_handler = workflow.execute_activity_method( + Activities.update_schedules, + { + 'schedules': schedule_actions['to_update'] + }, + retry_policy=retry_policy, + start_to_close_timeout=timedelta(seconds=60) + ) + + slot_deletion_report = await slot_deletion_report_handler + slot_insertion_report = await slot_insertion_report_handler + schedule_deletion_report = await schedule_deletion_report_handler + schedule_insertion_report = await schedule_insertion_report_handler + schedule_update_report = await schedule_update_report_handler + + schedule_report_handler = workflow.execute_activity_method( + Activities.report_schedule_orchestration, + { + 'created_schedules': schedule_insertion_report, + 'updated_schedules': schedule_update_report, + 'deleted_schedules': schedule_deletion_report + }, + retry_policy=retry_policy, + start_to_close_timeout=timedelta(seconds=60) + ) + + slot_report_handler = workflow.execute_activity_method( + Activities.report_slot_orchestration, + { + 'inserted_slots': slot_insertion_report, + 'deleted_slots': slot_deletion_report + }, + retry_policy=retry_policy, + start_to_close_timeout=timedelta(seconds=60) + ) + + await schedule_report_handler + await slot_report_handler diff --git a/test.ipynb b/test.ipynb index 8a0ebc6..9683267 100644 --- a/test.ipynb +++ b/test.ipynb @@ -136,19 +136,33 @@ }, { "cell_type": "code", - "execution_count": 81, + "execution_count": 2, "id": "bb750ae6", "metadata": {}, "outputs": [ { - "data": { - "text/plain": [ - "" - ] - }, - "execution_count": 81, - "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:meu-schedule-id5, RunId: 01971d9e-b19b-7e25-8f60-ba4ed95e12e4.', b'\\x08\\x06\\x12\\x88\\x01Workflow execution is already running. WorkflowId: temporal-sys-scheduler:meu-schedule-id5, RunId: 01971d9e-b19b-7e25-8f60-ba4ed95e12e4.\\x1a\\xa7\\x01\\nWtype.googleapis.com/temporal.api.errordetails.v1.WorkflowExecutionAlreadyStartedFailure\\x12L\\n$6ae80236-ccbf-45b5-8282-1ef14a559b59\\x12$01971d9e-b19b-7e25-8f60-ba4ed95e12e4')", + "\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:meu-schedule-id5, RunId: 01971d9e-b19b-7e25-8f60-ba4ed95e12e4.", + "\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[2]\u001b[39m\u001b[32m, line 17\u001b[39m\n\u001b[32m 13\u001b[39m customer_id_key = SearchAttributeKey.for_keyword(\u001b[33m\"\u001b[39m\u001b[33mOrchestrated\u001b[39m\u001b[33m\"\u001b[39m)\n\u001b[32m 14\u001b[39m search_attributes = TypedSearchAttributes([\n\u001b[32m 15\u001b[39m SearchAttributePair(customer_id_key, \u001b[33m\"\u001b[39m\u001b[33mtrue\u001b[39m\u001b[33m\"\u001b[39m)\n\u001b[32m 16\u001b[39m ])\n\u001b[32m---> \u001b[39m\u001b[32m17\u001b[39m \u001b[38;5;28;01mawait\u001b[39;00m temporal_client.create_schedule(\n\u001b[32m 18\u001b[39m \u001b[33m\"\u001b[39m\u001b[33mmeu-schedule-id5\u001b[39m\u001b[33m\"\u001b[39m,\n\u001b[32m 19\u001b[39m Schedule(\n\u001b[32m 20\u001b[39m action=ScheduleActionStartWorkflow(\n\u001b[32m 21\u001b[39m \u001b[33m'\u001b[39m\u001b[33mscouter-test2\u001b[39m\u001b[33m'\u001b[39m,\n\u001b[32m 22\u001b[39m {\n\u001b[32m 23\u001b[39m \u001b[33m'\u001b[39m\u001b[33margs\u001b[39m\u001b[33m'\u001b[39m: {\n\u001b[32m 24\u001b[39m \u001b[33m'\u001b[39m\u001b[33marg1\u001b[39m\u001b[33m'\u001b[39m: \u001b[33m'\u001b[39m\u001b[33mvalue1\u001b[39m\u001b[33m'\u001b[39m\n\u001b[32m 25\u001b[39m }\n\u001b[32m 26\u001b[39m },\n\u001b[32m 27\u001b[39m \u001b[38;5;28mid\u001b[39m=\u001b[33m\"\u001b[39m\u001b[33mworkflow-id-unico\u001b[39m\u001b[33m\"\u001b[39m,\n\u001b[32m 28\u001b[39m task_queue=\u001b[33m\"\u001b[39m\u001b[33mnome-da-task-queue\u001b[39m\u001b[33m\"\u001b[39m,\n\u001b[32m 29\u001b[39m ),\n\u001b[32m 30\u001b[39m spec=ScheduleSpec(\n\u001b[32m 31\u001b[39m intervals=[ScheduleIntervalSpec(every=timedelta(minutes=\u001b[32m10\u001b[39m))]\n\u001b[32m 32\u001b[39m )\n\u001b[32m 33\u001b[39m ),\n\u001b[32m 34\u001b[39m search_attributes=search_attributes,\n\u001b[32m 35\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": [ @@ -191,7 +205,7 @@ }, { "cell_type": "code", - "execution_count": 86, + "execution_count": 2, "id": "1bd82225", "metadata": {}, "outputs": [ @@ -200,474 +214,13 @@ "output_type": "stream", "text": [ "Getting orchestrated schedules...\n", - "Schedule: %s ScheduleListDescription(id='meu-schedule-id5', schedule=ScheduleListSchedule(action=ScheduleListActionStartWorkflow(workflow='scouter-test2'), spec=ScheduleSpec(calendars=[], intervals=[ScheduleIntervalSpec(every=datetime.timedelta(seconds=600), offset=None)], cron_expressions=[], skip=[], start_at=None, end_at=None, jitter=None, time_zone_name=None), state=ScheduleListState(note=None, paused=False)), info=ScheduleListInfo(recent_actions=[], next_action_times=[datetime.datetime(2025, 5, 29, 20, 0, tzinfo=datetime.timezone.utc), datetime.datetime(2025, 5, 29, 20, 10, tzinfo=datetime.timezone.utc), datetime.datetime(2025, 5, 29, 20, 20, tzinfo=datetime.timezone.utc), datetime.datetime(2025, 5, 29, 20, 30, tzinfo=datetime.timezone.utc), datetime.datetime(2025, 5, 29, 20, 40, tzinfo=datetime.timezone.utc)]), typed_search_attributes=TypedSearchAttributes(search_attributes=[SearchAttributePair(key=_SearchAttributeKey(_name='Orchestrated', _indexed_value_type=, _value_type=), value='true')]), search_attributes={'Orchestrated': ['true']}, data_converter=DataConverter(payload_converter_class=, payload_codec=None, failure_converter_class=, payload_converter=, failure_converter=), raw_entry=schedule_id: \"meu-schedule-id5\"\n", - "search_attributes {\n", - " indexed_fields {\n", - " key: \"Orchestrated\"\n", - " value {\n", - " metadata {\n", - " key: \"type\"\n", - " value: \"Text\"\n", - " }\n", - " metadata {\n", - " key: \"encoding\"\n", - " value: \"json/plain\"\n", - " }\n", - " data: \"\\\"true\\\"\"\n", - " }\n", - " }\n", - "}\n", - "info {\n", - " spec {\n", - " interval {\n", - " interval {\n", - " seconds: 600\n", - " }\n", - " }\n", - " }\n", - " workflow_type {\n", - " name: \"scouter-test2\"\n", - " }\n", - " future_action_times {\n", - " seconds: 1748548800\n", - " }\n", - " future_action_times {\n", - " seconds: 1748549400\n", - " }\n", - " future_action_times {\n", - " seconds: 1748550000\n", - " }\n", - " future_action_times {\n", - " seconds: 1748550600\n", - " }\n", - " future_action_times {\n", - " seconds: 1748551200\n", - " }\n", - "}\n", - ")\n", - "Search attributes: %s {'Orchestrated': ['true']}\n", - "Schedule: %s ScheduleListDescription(id='meu-schedule-id4', schedule=ScheduleListSchedule(action=ScheduleListActionStartWorkflow(workflow='scouter-test2'), spec=ScheduleSpec(calendars=[], intervals=[ScheduleIntervalSpec(every=datetime.timedelta(seconds=600), offset=None)], cron_expressions=[], skip=[], start_at=None, end_at=None, jitter=None, time_zone_name=None), state=ScheduleListState(note=None, paused=False)), info=ScheduleListInfo(recent_actions=[], next_action_times=[datetime.datetime(2025, 5, 29, 20, 0, tzinfo=datetime.timezone.utc), datetime.datetime(2025, 5, 29, 20, 10, tzinfo=datetime.timezone.utc), datetime.datetime(2025, 5, 29, 20, 20, tzinfo=datetime.timezone.utc), datetime.datetime(2025, 5, 29, 20, 30, tzinfo=datetime.timezone.utc), datetime.datetime(2025, 5, 29, 20, 40, tzinfo=datetime.timezone.utc)]), typed_search_attributes=TypedSearchAttributes(search_attributes=[SearchAttributePair(key=_SearchAttributeKey(_name='Orchestrated', _indexed_value_type=, _value_type=), value='true')]), search_attributes={'Orchestrated': ['true']}, data_converter=DataConverter(payload_converter_class=, payload_codec=None, failure_converter_class=, payload_converter=, failure_converter=), raw_entry=schedule_id: \"meu-schedule-id4\"\n", - "search_attributes {\n", - " indexed_fields {\n", - " key: \"Orchestrated\"\n", - " value {\n", - " metadata {\n", - " key: \"type\"\n", - " value: \"Text\"\n", - " }\n", - " metadata {\n", - " key: \"encoding\"\n", - " value: \"json/plain\"\n", - " }\n", - " data: \"\\\"true\\\"\"\n", - " }\n", - " }\n", - "}\n", - "info {\n", - " spec {\n", - " interval {\n", - " interval {\n", - " seconds: 600\n", - " }\n", - " }\n", - " }\n", - " workflow_type {\n", - " name: \"scouter-test2\"\n", - " }\n", - " future_action_times {\n", - " seconds: 1748548800\n", - " }\n", - " future_action_times {\n", - " seconds: 1748549400\n", - " }\n", - " future_action_times {\n", - " seconds: 1748550000\n", - " }\n", - " future_action_times {\n", - " seconds: 1748550600\n", - " }\n", - " future_action_times {\n", - " seconds: 1748551200\n", - " }\n", - "}\n", - ")\n", - "Search attributes: %s {'Orchestrated': ['true']}\n", - "Schedule: %s ScheduleListDescription(id='meu-schedule-id3', schedule=ScheduleListSchedule(action=ScheduleListActionStartWorkflow(workflow='scouter-test2'), spec=ScheduleSpec(calendars=[], intervals=[ScheduleIntervalSpec(every=datetime.timedelta(seconds=600), offset=None)], cron_expressions=[], skip=[], start_at=None, end_at=None, jitter=None, time_zone_name=None), state=ScheduleListState(note=None, paused=False)), info=ScheduleListInfo(recent_actions=[], next_action_times=[datetime.datetime(2025, 5, 29, 20, 0, tzinfo=datetime.timezone.utc), datetime.datetime(2025, 5, 29, 20, 10, tzinfo=datetime.timezone.utc), datetime.datetime(2025, 5, 29, 20, 20, tzinfo=datetime.timezone.utc), datetime.datetime(2025, 5, 29, 20, 30, tzinfo=datetime.timezone.utc), datetime.datetime(2025, 5, 29, 20, 40, tzinfo=datetime.timezone.utc)]), typed_search_attributes=TypedSearchAttributes(search_attributes=[SearchAttributePair(key=_SearchAttributeKey(_name='Orchestrated', _indexed_value_type=, _value_type=), value='true')]), search_attributes={'Orchestrated': ['true']}, data_converter=DataConverter(payload_converter_class=, payload_codec=None, failure_converter_class=, payload_converter=, failure_converter=), raw_entry=schedule_id: \"meu-schedule-id3\"\n", - "search_attributes {\n", - " indexed_fields {\n", - " key: \"Orchestrated\"\n", - " value {\n", - " metadata {\n", - " key: \"type\"\n", - " value: \"Text\"\n", - " }\n", - " metadata {\n", - " key: \"encoding\"\n", - " value: \"json/plain\"\n", - " }\n", - " data: \"\\\"true\\\"\"\n", - " }\n", - " }\n", - "}\n", - "info {\n", - " spec {\n", - " interval {\n", - " interval {\n", - " seconds: 600\n", - " }\n", - " }\n", - " }\n", - " workflow_type {\n", - " name: \"scouter-test2\"\n", - " }\n", - " future_action_times {\n", - " seconds: 1748548800\n", - " }\n", - " future_action_times {\n", - " seconds: 1748549400\n", - " }\n", - " future_action_times {\n", - " seconds: 1748550000\n", - " }\n", - " future_action_times {\n", - " seconds: 1748550600\n", - " }\n", - " future_action_times {\n", - " seconds: 1748551200\n", - " }\n", - "}\n", - ")\n", - "Search attributes: %s {'Orchestrated': ['true']}\n", - "Schedule: %s ScheduleListDescription(id='meu-schedule-id2', schedule=ScheduleListSchedule(action=ScheduleListActionStartWorkflow(workflow='scouter-test2'), spec=ScheduleSpec(calendars=[], intervals=[ScheduleIntervalSpec(every=datetime.timedelta(seconds=600), offset=None)], cron_expressions=[], skip=[], start_at=None, end_at=None, jitter=None, time_zone_name=None), state=ScheduleListState(note=None, paused=False)), info=ScheduleListInfo(recent_actions=[ScheduleActionResult(scheduled_at=datetime.datetime(2025, 5, 29, 19, 50, tzinfo=datetime.timezone.utc), started_at=datetime.datetime(2025, 5, 29, 19, 50, 0, 37623, tzinfo=datetime.timezone.utc), action=ScheduleActionExecutionStartWorkflow(workflow_id='workflow-id-unico-2025-05-29T19:50:00Z', first_execution_run_id='01971d98-265a-7285-b46f-cf09bcf2d301'))], next_action_times=[datetime.datetime(2025, 5, 29, 20, 0, tzinfo=datetime.timezone.utc), datetime.datetime(2025, 5, 29, 20, 10, tzinfo=datetime.timezone.utc), datetime.datetime(2025, 5, 29, 20, 20, tzinfo=datetime.timezone.utc), datetime.datetime(2025, 5, 29, 20, 30, tzinfo=datetime.timezone.utc), datetime.datetime(2025, 5, 29, 20, 40, tzinfo=datetime.timezone.utc)]), typed_search_attributes=TypedSearchAttributes(search_attributes=[SearchAttributePair(key=_SearchAttributeKey(_name='Orchestrated', _indexed_value_type=, _value_type=), value='true')]), search_attributes={'Orchestrated': ['true']}, data_converter=DataConverter(payload_converter_class=, payload_codec=None, failure_converter_class=, payload_converter=, failure_converter=), raw_entry=schedule_id: \"meu-schedule-id2\"\n", - "search_attributes {\n", - " indexed_fields {\n", - " key: \"Orchestrated\"\n", - " value {\n", - " metadata {\n", - " key: \"type\"\n", - " value: \"Text\"\n", - " }\n", - " metadata {\n", - " key: \"encoding\"\n", - " value: \"json/plain\"\n", - " }\n", - " data: \"\\\"true\\\"\"\n", - " }\n", - " }\n", - "}\n", - "info {\n", - " spec {\n", - " interval {\n", - " interval {\n", - " seconds: 600\n", - " }\n", - " }\n", - " }\n", - " workflow_type {\n", - " name: \"scouter-test2\"\n", - " }\n", - " recent_actions {\n", - " schedule_time {\n", - " seconds: 1748548200\n", - " }\n", - " actual_time {\n", - " seconds: 1748548200\n", - " nanos: 37623285\n", - " }\n", - " start_workflow_result {\n", - " workflow_id: \"workflow-id-unico-2025-05-29T19:50:00Z\"\n", - " run_id: \"01971d98-265a-7285-b46f-cf09bcf2d301\"\n", - " }\n", - " start_workflow_status: WORKFLOW_EXECUTION_STATUS_RUNNING\n", - " }\n", - " future_action_times {\n", - " seconds: 1748548800\n", - " }\n", - " future_action_times {\n", - " seconds: 1748549400\n", - " }\n", - " future_action_times {\n", - " seconds: 1748550000\n", - " }\n", - " future_action_times {\n", - " seconds: 1748550600\n", - " }\n", - " future_action_times {\n", - " seconds: 1748551200\n", - " }\n", - "}\n", - ")\n", - "Search attributes: %s {'Orchestrated': ['true']}\n", - "Schedule: %s ScheduleListDescription(id='meu-schedule-id', schedule=ScheduleListSchedule(action=ScheduleListActionStartWorkflow(workflow='scouter-test'), spec=ScheduleSpec(calendars=[], intervals=[ScheduleIntervalSpec(every=datetime.timedelta(seconds=600), offset=None)], cron_expressions=[], skip=[], start_at=None, end_at=None, jitter=None, time_zone_name=None), state=ScheduleListState(note=None, paused=False)), info=ScheduleListInfo(recent_actions=[ScheduleActionResult(scheduled_at=datetime.datetime(2025, 5, 29, 19, 0, tzinfo=datetime.timezone.utc), started_at=datetime.datetime(2025, 5, 29, 19, 0, 0, 37788, tzinfo=datetime.timezone.utc), action=ScheduleActionExecutionStartWorkflow(workflow_id='workflow-id-unico-2025-05-29T19:00:00Z', first_execution_run_id='01971d6a-5fa1-70f8-8371-60ad11587b77'))], next_action_times=[datetime.datetime(2025, 5, 29, 20, 0, tzinfo=datetime.timezone.utc), datetime.datetime(2025, 5, 29, 20, 10, tzinfo=datetime.timezone.utc), datetime.datetime(2025, 5, 29, 20, 20, tzinfo=datetime.timezone.utc), datetime.datetime(2025, 5, 29, 20, 30, tzinfo=datetime.timezone.utc), datetime.datetime(2025, 5, 29, 20, 40, tzinfo=datetime.timezone.utc)]), typed_search_attributes=TypedSearchAttributes(search_attributes=[SearchAttributePair(key=_SearchAttributeKey(_name='Orchestrated', _indexed_value_type=, _value_type=), value='true')]), search_attributes={'Orchestrated': ['true']}, data_converter=DataConverter(payload_converter_class=, payload_codec=None, failure_converter_class=, payload_converter=, failure_converter=), raw_entry=schedule_id: \"meu-schedule-id\"\n", - "search_attributes {\n", - " indexed_fields {\n", - " key: \"Orchestrated\"\n", - " value {\n", - " metadata {\n", - " key: \"type\"\n", - " value: \"Text\"\n", - " }\n", - " metadata {\n", - " key: \"encoding\"\n", - " value: \"json/plain\"\n", - " }\n", - " data: \"\\\"true\\\"\"\n", - " }\n", - " }\n", - "}\n", - "info {\n", - " spec {\n", - " interval {\n", - " interval {\n", - " seconds: 600\n", - " }\n", - " }\n", - " }\n", - " workflow_type {\n", - " name: \"scouter-test\"\n", - " }\n", - " recent_actions {\n", - " schedule_time {\n", - " seconds: 1748545200\n", - " }\n", - " actual_time {\n", - " seconds: 1748545200\n", - " nanos: 37788631\n", - " }\n", - " start_workflow_result {\n", - " workflow_id: \"workflow-id-unico-2025-05-29T19:00:00Z\"\n", - " run_id: \"01971d6a-5fa1-70f8-8371-60ad11587b77\"\n", - " }\n", - " start_workflow_status: WORKFLOW_EXECUTION_STATUS_RUNNING\n", - " }\n", - " future_action_times {\n", - " seconds: 1748548800\n", - " }\n", - " future_action_times {\n", - " seconds: 1748549400\n", - " }\n", - " future_action_times {\n", - " seconds: 1748550000\n", - " }\n", - " future_action_times {\n", - " seconds: 1748550600\n", - " }\n", - " future_action_times {\n", - " seconds: 1748551200\n", - " }\n", - "}\n", - ")\n", - "Search attributes: %s {'Orchestrated': ['true']}\n", - "Schedule: %s ScheduleListDescription(id='laborious_test', schedule=ScheduleListSchedule(action=ScheduleListActionStartWorkflow(workflow='predictions_batch'), spec=ScheduleSpec(calendars=[], intervals=[ScheduleIntervalSpec(every=datetime.timedelta(seconds=60), offset=datetime.timedelta(0))], cron_expressions=[], skip=[], start_at=None, end_at=None, jitter=None, time_zone_name=None), state=ScheduleListState(note=None, paused=False)), info=ScheduleListInfo(recent_actions=[ScheduleActionResult(scheduled_at=datetime.datetime(2025, 5, 29, 19, 53, tzinfo=datetime.timezone.utc), started_at=datetime.datetime(2025, 5, 29, 19, 53, 0, 35750, tzinfo=datetime.timezone.utc), action=ScheduleActionExecutionStartWorkflow(workflow_id='laborious_test-2025-05-29T19:53:00Z', first_execution_run_id='01971d9a-e57f-700b-953b-bdb3b05810e2')), ScheduleActionResult(scheduled_at=datetime.datetime(2025, 5, 29, 19, 54, tzinfo=datetime.timezone.utc), started_at=datetime.datetime(2025, 5, 29, 19, 54, 0, 35780, tzinfo=datetime.timezone.utc), action=ScheduleActionExecutionStartWorkflow(workflow_id='laborious_test-2025-05-29T19:54:00Z', first_execution_run_id='01971d9b-cfde-7f39-8b38-b7e62fee1f82')), ScheduleActionResult(scheduled_at=datetime.datetime(2025, 5, 29, 19, 55, tzinfo=datetime.timezone.utc), started_at=datetime.datetime(2025, 5, 29, 19, 55, 0, 34329, tzinfo=datetime.timezone.utc), action=ScheduleActionExecutionStartWorkflow(workflow_id='laborious_test-2025-05-29T19:55:00Z', first_execution_run_id='01971d9c-ba3c-7d27-9db7-72759ebabc76')), ScheduleActionResult(scheduled_at=datetime.datetime(2025, 5, 29, 19, 56, tzinfo=datetime.timezone.utc), started_at=datetime.datetime(2025, 5, 29, 19, 56, 0, 49214, tzinfo=datetime.timezone.utc), action=ScheduleActionExecutionStartWorkflow(workflow_id='laborious_test-2025-05-29T19:56:00Z', first_execution_run_id='01971d9d-a4a9-7b55-a77b-fd450e768ccf')), ScheduleActionResult(scheduled_at=datetime.datetime(2025, 5, 29, 19, 57, tzinfo=datetime.timezone.utc), started_at=datetime.datetime(2025, 5, 29, 19, 57, 0, 36109, tzinfo=datetime.timezone.utc), action=ScheduleActionExecutionStartWorkflow(workflow_id='laborious_test-2025-05-29T19:57:00Z', first_execution_run_id='01971d9e-8eff-7608-9b10-0ed1875df9fe'))], next_action_times=[datetime.datetime(2025, 5, 29, 19, 58, tzinfo=datetime.timezone.utc), datetime.datetime(2025, 5, 29, 19, 59, tzinfo=datetime.timezone.utc), datetime.datetime(2025, 5, 29, 20, 0, tzinfo=datetime.timezone.utc), datetime.datetime(2025, 5, 29, 20, 1, tzinfo=datetime.timezone.utc), datetime.datetime(2025, 5, 29, 20, 2, tzinfo=datetime.timezone.utc)]), typed_search_attributes=TypedSearchAttributes(search_attributes=[]), search_attributes={}, data_converter=DataConverter(payload_converter_class=, payload_codec=None, failure_converter_class=, payload_converter=, failure_converter=), raw_entry=schedule_id: \"laborious_test\"\n", - "info {\n", - " spec {\n", - " interval {\n", - " interval {\n", - " seconds: 60\n", - " }\n", - " phase {\n", - " }\n", - " }\n", - " }\n", - " workflow_type {\n", - " name: \"predictions_batch\"\n", - " }\n", - " recent_actions {\n", - " schedule_time {\n", - " seconds: 1748548380\n", - " }\n", - " actual_time {\n", - " seconds: 1748548380\n", - " nanos: 35750962\n", - " }\n", - " start_workflow_result {\n", - " workflow_id: \"laborious_test-2025-05-29T19:53:00Z\"\n", - " run_id: \"01971d9a-e57f-700b-953b-bdb3b05810e2\"\n", - " }\n", - " start_workflow_status: WORKFLOW_EXECUTION_STATUS_COMPLETED\n", - " }\n", - " recent_actions {\n", - " schedule_time {\n", - " seconds: 1748548440\n", - " }\n", - " actual_time {\n", - " seconds: 1748548440\n", - " nanos: 35780414\n", - " }\n", - " start_workflow_result {\n", - " workflow_id: \"laborious_test-2025-05-29T19:54:00Z\"\n", - " run_id: \"01971d9b-cfde-7f39-8b38-b7e62fee1f82\"\n", - " }\n", - " start_workflow_status: WORKFLOW_EXECUTION_STATUS_COMPLETED\n", - " }\n", - " recent_actions {\n", - " schedule_time {\n", - " seconds: 1748548500\n", - " }\n", - " actual_time {\n", - " seconds: 1748548500\n", - " nanos: 34329717\n", - " }\n", - " start_workflow_result {\n", - " workflow_id: \"laborious_test-2025-05-29T19:55:00Z\"\n", - " run_id: \"01971d9c-ba3c-7d27-9db7-72759ebabc76\"\n", - " }\n", - " start_workflow_status: WORKFLOW_EXECUTION_STATUS_COMPLETED\n", - " }\n", - " recent_actions {\n", - " schedule_time {\n", - " seconds: 1748548560\n", - " }\n", - " actual_time {\n", - " seconds: 1748548560\n", - " nanos: 49214684\n", - " }\n", - " start_workflow_result {\n", - " workflow_id: \"laborious_test-2025-05-29T19:56:00Z\"\n", - " run_id: \"01971d9d-a4a9-7b55-a77b-fd450e768ccf\"\n", - " }\n", - " start_workflow_status: WORKFLOW_EXECUTION_STATUS_COMPLETED\n", - " }\n", - " recent_actions {\n", - " schedule_time {\n", - " seconds: 1748548620\n", - " }\n", - " actual_time {\n", - " seconds: 1748548620\n", - " nanos: 36109875\n", - " }\n", - " start_workflow_result {\n", - " workflow_id: \"laborious_test-2025-05-29T19:57:00Z\"\n", - " run_id: \"01971d9e-8eff-7608-9b10-0ed1875df9fe\"\n", - " }\n", - " start_workflow_status: WORKFLOW_EXECUTION_STATUS_RUNNING\n", - " }\n", - " future_action_times {\n", - " seconds: 1748548680\n", - " }\n", - " future_action_times {\n", - " seconds: 1748548740\n", - " }\n", - " future_action_times {\n", - " seconds: 1748548800\n", - " }\n", - " future_action_times {\n", - " seconds: 1748548860\n", - " }\n", - " future_action_times {\n", - " seconds: 1748548920\n", - " }\n", - "}\n", - ")\n", - "Search attributes: %s {}\n", - "Schedule: %s ScheduleListDescription(id='scouter-opcua-pipeline', schedule=ScheduleListSchedule(action=ScheduleListActionStartWorkflow(workflow='scouter'), spec=ScheduleSpec(calendars=[], intervals=[ScheduleIntervalSpec(every=datetime.timedelta(seconds=30), offset=datetime.timedelta(0))], cron_expressions=[], skip=[], start_at=None, end_at=None, jitter=None, time_zone_name=None), state=ScheduleListState(note=None, paused=False)), info=ScheduleListInfo(recent_actions=[ScheduleActionResult(scheduled_at=datetime.datetime(2025, 5, 29, 19, 55, tzinfo=datetime.timezone.utc), started_at=datetime.datetime(2025, 5, 29, 19, 55, 0, 26130, tzinfo=datetime.timezone.utc), action=ScheduleActionExecutionStartWorkflow(workflow_id='scouter-opcua-pipeline-2025-05-29T19:55:00Z', first_execution_run_id='01971d9c-ba35-7b2a-b596-effe9939c7c7')), ScheduleActionResult(scheduled_at=datetime.datetime(2025, 5, 29, 19, 55, 30, tzinfo=datetime.timezone.utc), started_at=datetime.datetime(2025, 5, 29, 19, 55, 30, 26550, tzinfo=datetime.timezone.utc), action=ScheduleActionExecutionStartWorkflow(workflow_id='scouter-opcua-pipeline-2025-05-29T19:55:30Z', first_execution_run_id='01971d9d-2f66-7012-bd77-e5809b8c4c7a')), ScheduleActionResult(scheduled_at=datetime.datetime(2025, 5, 29, 19, 56, tzinfo=datetime.timezone.utc), started_at=datetime.datetime(2025, 5, 29, 19, 56, 0, 39848, tzinfo=datetime.timezone.utc), action=ScheduleActionExecutionStartWorkflow(workflow_id='scouter-opcua-pipeline-2025-05-29T19:56:00Z', first_execution_run_id='01971d9d-a4a1-7ba9-9205-21f157fe6e54')), ScheduleActionResult(scheduled_at=datetime.datetime(2025, 5, 29, 19, 56, 30, tzinfo=datetime.timezone.utc), started_at=datetime.datetime(2025, 5, 29, 19, 56, 30, 39791, tzinfo=datetime.timezone.utc), action=ScheduleActionExecutionStartWorkflow(workflow_id='scouter-opcua-pipeline-2025-05-29T19:56:30Z', first_execution_run_id='01971d9e-19d2-7358-96d3-261e203faaa7')), ScheduleActionResult(scheduled_at=datetime.datetime(2025, 5, 29, 19, 57, tzinfo=datetime.timezone.utc), started_at=datetime.datetime(2025, 5, 29, 19, 57, 0, 28221, tzinfo=datetime.timezone.utc), action=ScheduleActionExecutionStartWorkflow(workflow_id='scouter-opcua-pipeline-2025-05-29T19:57:00Z', first_execution_run_id='01971d9e-8ef8-72af-a903-1391b5e06c2f'))], next_action_times=[datetime.datetime(2025, 5, 29, 19, 57, 30, tzinfo=datetime.timezone.utc), datetime.datetime(2025, 5, 29, 19, 58, tzinfo=datetime.timezone.utc), datetime.datetime(2025, 5, 29, 19, 58, 30, tzinfo=datetime.timezone.utc), datetime.datetime(2025, 5, 29, 19, 59, tzinfo=datetime.timezone.utc), datetime.datetime(2025, 5, 29, 19, 59, 30, tzinfo=datetime.timezone.utc)]), typed_search_attributes=TypedSearchAttributes(search_attributes=[]), search_attributes={}, data_converter=DataConverter(payload_converter_class=, payload_codec=None, failure_converter_class=, payload_converter=, failure_converter=), raw_entry=schedule_id: \"scouter-opcua-pipeline\"\n", - "info {\n", - " spec {\n", - " interval {\n", - " interval {\n", - " seconds: 30\n", - " }\n", - " phase {\n", - " }\n", - " }\n", - " }\n", - " workflow_type {\n", - " name: \"scouter\"\n", - " }\n", - " recent_actions {\n", - " schedule_time {\n", - " seconds: 1748548500\n", - " }\n", - " actual_time {\n", - " seconds: 1748548500\n", - " nanos: 26130839\n", - " }\n", - " start_workflow_result {\n", - " workflow_id: \"scouter-opcua-pipeline-2025-05-29T19:55:00Z\"\n", - " run_id: \"01971d9c-ba35-7b2a-b596-effe9939c7c7\"\n", - " }\n", - " start_workflow_status: WORKFLOW_EXECUTION_STATUS_COMPLETED\n", - " }\n", - " recent_actions {\n", - " schedule_time {\n", - " seconds: 1748548530\n", - " }\n", - " actual_time {\n", - " seconds: 1748548530\n", - " nanos: 26550164\n", - " }\n", - " start_workflow_result {\n", - " workflow_id: \"scouter-opcua-pipeline-2025-05-29T19:55:30Z\"\n", - " run_id: \"01971d9d-2f66-7012-bd77-e5809b8c4c7a\"\n", - " }\n", - " start_workflow_status: WORKFLOW_EXECUTION_STATUS_COMPLETED\n", - " }\n", - " recent_actions {\n", - " schedule_time {\n", - " seconds: 1748548560\n", - " }\n", - " actual_time {\n", - " seconds: 1748548560\n", - " nanos: 39848794\n", - " }\n", - " start_workflow_result {\n", - " workflow_id: \"scouter-opcua-pipeline-2025-05-29T19:56:00Z\"\n", - " run_id: \"01971d9d-a4a1-7ba9-9205-21f157fe6e54\"\n", - " }\n", - " start_workflow_status: WORKFLOW_EXECUTION_STATUS_COMPLETED\n", - " }\n", - " recent_actions {\n", - " schedule_time {\n", - " seconds: 1748548590\n", - " }\n", - " actual_time {\n", - " seconds: 1748548590\n", - " nanos: 39791709\n", - " }\n", - " start_workflow_result {\n", - " workflow_id: \"scouter-opcua-pipeline-2025-05-29T19:56:30Z\"\n", - " run_id: \"01971d9e-19d2-7358-96d3-261e203faaa7\"\n", - " }\n", - " start_workflow_status: WORKFLOW_EXECUTION_STATUS_COMPLETED\n", - " }\n", - " recent_actions {\n", - " schedule_time {\n", - " seconds: 1748548620\n", - " }\n", - " actual_time {\n", - " seconds: 1748548620\n", - " nanos: 28221068\n", - " }\n", - " start_workflow_result {\n", - " workflow_id: \"scouter-opcua-pipeline-2025-05-29T19:57:00Z\"\n", - " run_id: \"01971d9e-8ef8-72af-a903-1391b5e06c2f\"\n", - " }\n", - " start_workflow_status: WORKFLOW_EXECUTION_STATUS_RUNNING\n", - " }\n", - " future_action_times {\n", - " seconds: 1748548650\n", - " }\n", - " future_action_times {\n", - " seconds: 1748548680\n", - " }\n", - " future_action_times {\n", - " seconds: 1748548710\n", - " }\n", - " future_action_times {\n", - " seconds: 1748548740\n", - " }\n", - " future_action_times {\n", - " seconds: 1748548770\n", - " }\n", - "}\n", - ")\n", - "Search attributes: %s {}\n", - "Found %d orchestrated schedules 5\n" + "Found %d orchestrated schedules 5\n", + "Orchestrated schedules: %s {'meu-schedule-id5': {'frequency': 600, 'data': {'args': {'arg1': 'value1'}}, 'handle': }, 'meu-schedule-id4': {'frequency': 600, 'data': {}, 'handle': }, 'meu-schedule-id3': {'frequency': 600, 'data': {}, 'handle': }, 'meu-schedule-id2': {'frequency': 600, 'data': {}, 'handle': }, 'meu-schedule-id': {'frequency': 600, 'data': {}, 'handle': }}\n" ] } ], "source": [ - "schedules = await manager.load_schedule({})" + "schedules = await manager.load_schedule()" ] }, { diff --git a/tests/orchestrator/activities/test_activities.py b/tests/orchestrator/activities/test_activities.py index 14d857c..997a55a 100644 --- a/tests/orchestrator/activities/test_activities.py +++ b/tests/orchestrator/activities/test_activities.py @@ -4,12 +4,15 @@ from orchestrator.activities.activities import Activities from orchestrator.activities.couchbase import Couchbase from orchestrator.activities.temporal_manager import TemporalManager from orchestrator.activities.slot_manager import SlotManager +from orchestrator.activities.formatters import Formatters @patch('orchestrator.activities.couchbase.Couchbase.__init__') @patch('orchestrator.activities.temporal_manager.TemporalManager.__init__') @patch('orchestrator.activities.slot_manager.SlotManager.__init__') -def test___init__(mock_slot_manager_init, mock_temporal_manager_init, +@patch('orchestrator.activities.formatters.Formatters.__init__') +def test___init__(mock_formatters_init, mock_slot_manager_init, + mock_temporal_manager_init, mock_couchbase_init): couchbase_config = { @@ -41,6 +44,7 @@ def test___init__(mock_slot_manager_init, mock_temporal_manager_init, assert isinstance(activities, Couchbase) assert isinstance(activities, TemporalManager) assert isinstance(activities, SlotManager) + assert isinstance(activities, Formatters) mock_slot_manager_init.assert_called_once_with( ANY, @@ -68,6 +72,12 @@ def test___init__(mock_slot_manager_init, mock_temporal_manager_init, notification_handler=notification_handler ) + mock_formatters_init.assert_called_once_with( + ANY, + logger=logger, + notification_handler=notification_handler + ) + @mark.asyncio @patch('orchestrator.activities.couchbase.Cluster') diff --git a/tests/orchestrator/activities/test_formatters.py b/tests/orchestrator/activities/test_formatters.py new file mode 100644 index 0000000..6ba74b9 --- /dev/null +++ b/tests/orchestrator/activities/test_formatters.py @@ -0,0 +1,480 @@ +from unittest.mock import MagicMock, patch, call, ANY +import json +from pytest import fixture, mark +from sientia_do.notifications.models import NotificationLevel +from orchestrator.activities.formatters import Formatters +from orchestrator.utils.orchestrator_functions import build_tag_config + + +@fixture +def formatters(): + return Formatters( + logger=MagicMock(), + notification_handler=MagicMock() + ) + + +@mark.asyncio +@patch("orchestrator.activities.formatters.scouter", + return_value="test_scouter") +@patch("orchestrator.activities.formatters.predictions_batch", + return_value="test_predictions_batch") +async def test_process_schedules(mock_predictions_batch, mock_scouter, formatters): + input_data = { + "pipelines": [ + { + "schedule_name": "test_schedule_name", + "workflow_type": "scouter", + "model_name": "test_model_name", + "model_id": "test_model_id" + }, + { + "schedule_name": "test_schedule_name2", + "workflow_type": "predictions_batch", + "model_name": "test_model_name", + "model_id": "test_model_id" + } + ] + } + + result = await formatters.process_schedules(input_data) + + assert result == { + "test_schedule_name": "test_scouter", + "test_schedule_name2": "test_predictions_batch" + } + + mock_scouter.assert_called_once_with(input_data['pipelines'][0]) + mock_predictions_batch.assert_called_once_with(input_data['pipelines'][1]) + + +@mark.asyncio +@patch("orchestrator.activities.formatters.gather_read_tags", + return_value={ + "test_server_name:test_tag_address": { + "server_name": "test_server_name", + "tag_address": "test_tag_address", + "topics": ["raw_test_schedule"] + }, + "test_server_name2:test_tag_address2": { + "server_name": "test_server_name2", + "tag_address": "test_tag_address2", + "topics": ["raw_test_schedule2"] + }, + "test_server_name2:test_tag_address3": { + "server_name": "test_server_name2", + "tag_address": "test_tag_address3", + "topics": ["raw_test_schedule2"] + } + }) +@patch("orchestrator.activities.formatters.build_tag_config", side_effect=build_tag_config) +async def test_process_slots(mock_build_tag_config, mock_gather_read_tags, formatters): + input_data = { + "opc_servers": [ + { + "server_name": "test_server_name", + "url": "test_url", + "uri": "test_uri", + "security_spec": { + "test_name": "test_spec" + } + }, + { + "server_name": "test_server_name2", + "url": "test_url2", + "uri": "test_uri2" + } + ], + "active_ingestors": [ + "test_active_ingestor1", + "test_active_ingestor2" + ], + "pipelines": "test_gather_read_tags" + } + + result = await formatters.process_slots(input_data) + + mock_gather_read_tags.assert_called_once_with(input_data['pipelines']) + mock_build_tag_config.assert_has_calls([ + call( + { + "server_name": "test_server_name", + "tag_address": "test_tag_address", + "topics": ["raw_test_schedule"] + }, + ANY, + { + "test_server_name": { + "server_name": "test_server_name", + "url": "test_url", + "uri": "test_uri", + "security_spec": { + "test_name": "test_spec" + } + }, + "test_server_name2": { + "server_name": "test_server_name2", + "url": "test_url2", + "uri": "test_uri2" + } + }, + 1 + ) + ]) + mock_build_tag_config.assert_has_calls([ + call( + { + "server_name": "test_server_name2", + "tag_address": "test_tag_address2", + "topics": ["raw_test_schedule2"] + }, + ANY, + { + "test_server_name": { + "server_name": "test_server_name", + "url": "test_url", + "uri": "test_uri", + "security_spec": { + "test_name": "test_spec" + } + }, + "test_server_name2": { + "server_name": "test_server_name2", + "url": "test_url2", + "uri": "test_uri2" + } + }, + 1 + ) + ]) + mock_build_tag_config.assert_has_calls([ + call( + { + "server_name": "test_server_name2", + "tag_address": "test_tag_address3", + "topics": ["raw_test_schedule2"] + }, + ANY, + { + "test_server_name": { + "server_name": "test_server_name", + "url": "test_url", + "uri": "test_uri", + "security_spec": { + "test_name": "test_spec" + } + }, + "test_server_name2": { + "server_name": "test_server_name2", + "url": "test_url2", + "uri": "test_uri2" + } + }, + 2 + ) + ]) + + assert result == { + "1": { + "test_server_name": { + "name": "test_server_name", + "url": "test_url", + "server_uri": "test_uri", + "test_name": "test_spec", + "tags": { + "test_tag_address": { + "server_name": "test_server_name", + "tag_address": "test_tag_address", + "topics": ["raw_test_schedule"] + } + } + }, + "test_server_name2": { + "name": "test_server_name2", + "url": "test_url2", + "server_uri": "test_uri2", + "tags": { + "test_tag_address2": { + "server_name": "test_server_name2", + "tag_address": "test_tag_address2", + "topics": ["raw_test_schedule2"] + } + } + } + }, + "2": { + "test_server_name2": { + "name": "test_server_name2", + "url": "test_url2", + "server_uri": "test_uri2", + "tags": { + "test_tag_address3": { + "server_name": "test_server_name2", + "tag_address": "test_tag_address3", + "topics": ["raw_test_schedule2"] + } + } + } + } + } + + +@mark.asyncio +async def test_create_schedule_config(formatters): + input_data = { + "current_schedule_config": { + "test_schedule_name_to_delete": { + "frequency": 60, + "data": {"test": "test"} + }, + "test_schedule_name_to_update": { + "frequency": 60, + "data": {"test": "test"} + } + }, + "schedule_config": { + "test_schedule_name_to_create": { + "frequency": 60, + "data": {"test": "test"} + }, + "test_schedule_name_to_update": { + "frequency": 60, + "data": {"test": "test2"} + } + } + } + + result = await formatters.create_schedule_config(input_data) + + assert result == { + "to_create": { + "test_schedule_name_to_create": { + "frequency": 60, + "data": {"test": "test"} + } + }, + "to_update": { + "test_schedule_name_to_update": { + "frequency": 60, + "data": {"test": "test2"} + } + }, + "to_delete": [ + "test_schedule_name_to_delete" + ] + } + + +@mark.asyncio +async def test_create_slot_config(formatters): + input_data = { + "current_slot_config": { + "1": { + "frequency": 60, + "data": {"test": "test"} + }, + "2": { + "frequency": 60, + "data": {"test": "test"} + } + }, + "slot_config": { + "1": { + "frequency": 60, + "data": {"test": "test2"} + } + } + } + + result = await formatters.create_slot_config(input_data) + + assert result == { + "to_delete": [ + "2" + ], + "to_insert": { + "1": { + "frequency": 60, + "data": {"test": "test2"} + } + } + } + + +def test_send_success_report(formatters): + formatters.send_success_report("test_message", "test_notification_id") + formatters.notification_handler.build_and_send_notification.assert_called_once_with( + "test_notification_id", + "test_message", + "report_orchestration", + NotificationLevel.INFO + ) + + +def test_send_error_report(formatters): + formatters.send_error_report( + "test_message", "test_notification_id", {"test": "test"}) + formatters.notification_handler.build_and_send_notification.assert_called_once_with( + "test_notification_id", + "test_message", + "report_orchestration", + NotificationLevel.ERROR, + attachment_content=json.dumps( + {"test": "test"}, indent=4, sort_keys=True) + ) + + +def test_parse_report(formatters): + input_data = { + "test_schedule_name_to_create": { + "success": True + }, + "test_schedule_name_to_create_error": { + "success": False, + "error": "test_error" + } + } + + result = formatters.parse_report(input_data) + + assert result == ( + ["test_schedule_name_to_create"], + ["test_schedule_name_to_create_error"] + ) + + +@mark.asyncio +async def test_report_schedule_orchestration(formatters): + formatters.parse_report = MagicMock( + side_effect=formatters.parse_report + ) + formatters.send_success_report = MagicMock() + formatters.send_error_report = MagicMock() + + input_data = { + "created_schedules": { + "test_schedule_name_to_create": { + "success": True + }, + "test_schedule_name_to_create_error": { + "success": False, + "error": "test_error" + } + }, + "updated_schedules": { + "test_schedule_name_to_update": { + "success": True + }, + "test_schedule_name_to_update_error": { + "success": False, + "error": "test_error" + } + }, + "deleted_schedules": { + "test_schedule_name_to_delete": { + "success": True + }, + "test_schedule_name_to_delete_error": { + "success": False, + "error": "test_error" + } + } + } + + await formatters.report_schedule_orchestration(input_data) + + formatters.parse_report.assert_has_calls([ + call(input_data['created_schedules']), + call(input_data['updated_schedules']), + call(input_data['deleted_schedules']) + ]) + formatters.send_success_report.assert_has_calls([ + call( + "Created schedules: \n test_schedule_name_to_create", + "REPORT_ORCHESTRATION_CREATED_SCHEDULES" + ), + call( + "Updated schedules: \n test_schedule_name_to_update", + "REPORT_ORCHESTRATION_UPDATED_SCHEDULES" + ), + call( + "Deleted schedules: \n test_schedule_name_to_delete", + "REPORT_ORCHESTRATION_DELETED_SCHEDULES" + ) + ]) + formatters.send_error_report.assert_has_calls([ + call( + "Failed to create schedules: \n test_schedule_name_to_create_error", + "REPORT_ORCHESTRATION_CREATED_SCHEDULES", + input_data['created_schedules'] + ), + call( + "Failed to update schedules: \n test_schedule_name_to_update_error", + "REPORT_ORCHESTRATION_UPDATED_SCHEDULES", + input_data['updated_schedules'] + ), + call( + "Failed to delete schedules: \n test_schedule_name_to_delete_error", + "REPORT_ORCHESTRATION_DELETED_SCHEDULES", + input_data['deleted_schedules'] + ) + ]) + + +@mark.asyncio +async def test_report_slot_orchestration(formatters): + formatters.parse_report = MagicMock( + side_effect=formatters.parse_report + ) + formatters.send_success_report = MagicMock() + formatters.send_error_report = MagicMock() + + input_data = { + "inserted_slots": { + "test_slot_name_to_create": { + "success": True + }, + "test_slot_name_to_create_error": { + "success": False, + "error": "test_error" + } + }, + "deleted_slots": { + "test_slot_name_to_delete": { + "success": True + }, + "test_slot_name_to_delete_error": { + "success": False, + "error": "test_error" + } + } + } + + await formatters.report_slot_orchestration(input_data) + + formatters.parse_report.assert_has_calls([ + call(input_data['inserted_slots']), + call(input_data['deleted_slots']) + ]) + formatters.send_success_report.assert_has_calls([ + call( + "Inserted slots: \n test_slot_name_to_create", + "REPORT_ORCHESTRATION_INSERTED_SLOTS" + ), + call( + "Deleted slots: \n test_slot_name_to_delete", + "REPORT_ORCHESTRATION_DELETED_SLOTS" + ) + ]) + formatters.send_error_report.assert_has_calls([ + call( + "Failed to insert slots: \n test_slot_name_to_create_error", + "REPORT_ORCHESTRATION_INSERTED_SLOTS", + input_data['inserted_slots'] + ), + call( + "Failed to delete slots: \n test_slot_name_to_delete_error", + "REPORT_ORCHESTRATION_DELETED_SLOTS", + input_data['deleted_slots'] + ) + ]) diff --git a/tests/orchestrator/activities/test_slot_manager.py b/tests/orchestrator/activities/test_slot_manager.py index cca97ca..75919d4 100644 --- a/tests/orchestrator/activities/test_slot_manager.py +++ b/tests/orchestrator/activities/test_slot_manager.py @@ -1,4 +1,4 @@ -from unittest.mock import MagicMock, patch +from unittest.mock import MagicMock, patch, call from pytest import mark, fixture from orchestrator.activities.slot_manager import SlotManager @@ -55,3 +55,66 @@ async def test_load_active_ingestors(slot_manager): assert response == ["heartbeat:ingestor:1", "heartbeat:ingestor:2", "heartbeat:ingestor:3"] + + +@mark.asyncio +async def test_update_slots(slot_manager): + slot_manager.set = MagicMock( + side_effect=[ + None, + Exception("Test exception") + ] + ) + + response = await slot_manager.update_slots({ + "to_insert": { + "1": "value1", + "2": "value2" + } + }) + + slot_manager.set.assert_has_calls([ + call("slot:opc_tags:1", "value1", ttl=None), + call("slot:opc_tags:2", "value2", ttl=None) + ]) + + assert response == { + "1": { + "success": True, + "message": "Slot updated successfully" + }, + "2": { + "success": False, + "message": "Test exception" + } + } + + +@mark.asyncio +async def test_delete_slots(slot_manager): + slot_manager.redis_client.delete = MagicMock( + side_effect=[ + None, + Exception("Test exception") + ] + ) + + response = await slot_manager.delete_slots({ + "to_delete": ["1", "2"] + }) + + slot_manager.redis_client.delete.assert_has_calls([ + call("slot:opc_tags:1"), + call("slot:opc_tags:2") + ]) + + assert response == { + "1": { + "success": True, + "message": "Slot deleted successfully" + }, + "2": { + "success": False, + "message": "Test exception" + } + } diff --git a/tests/orchestrator/activities/test_temporal_manager.py b/tests/orchestrator/activities/test_temporal_manager.py index 6808bdc..810e562 100644 --- a/tests/orchestrator/activities/test_temporal_manager.py +++ b/tests/orchestrator/activities/test_temporal_manager.py @@ -1,8 +1,10 @@ -from unittest.mock import MagicMock, patch, AsyncMock +from unittest.mock import MagicMock, patch, AsyncMock, call +from datetime import timedelta import base64 import json from pytest import fixture, mark from orchestrator.activities.temporal_manager import TemporalManager +from orchestrator.utils.converters import parse_frequency @fixture @@ -41,7 +43,7 @@ async def test_load_schedule(_mock_message_to_dict, temporal_manager): temporal_manager.temporal_client.list_schedules = AsyncMock( return_value=async_iter()) - temporal_manager.temporal_client.get_schedule.return_value = MagicMock( + temporal_manager.temporal_client.get_schedule_handle.return_value = MagicMock( describe=AsyncMock( return_value=MagicMock( schedule=MagicMock( @@ -57,7 +59,7 @@ async def test_load_schedule(_mock_message_to_dict, temporal_manager): ) ) ) - temporal_manager.temporal_client.get_schedule.return_value.describe \ + temporal_manager.temporal_client.get_schedule_handle.return_value.describe \ .return_value.schedule.spec = MagicMock( intervals=[ MagicMock( @@ -74,7 +76,205 @@ async def test_load_schedule(_mock_message_to_dict, temporal_manager): assert response == { "test-schedule-id": { "frequency": 60, - "data": {"test": "test"}, - "handle": temporal_manager.temporal_client.get_schedule.return_value + "data": {"test": "test"} + } + } + + +@mark.asyncio +@patch("orchestrator.activities.temporal_manager.parse_frequency", + side_effect=parse_frequency) +@patch("orchestrator.activities.temporal_manager.Schedule") +@patch("orchestrator.activities.temporal_manager.ScheduleActionStartWorkflow") +@patch("orchestrator.activities.temporal_manager.ScheduleIntervalSpec") +@patch("orchestrator.activities.temporal_manager.ScheduleSpec") +@patch("orchestrator.activities.temporal_manager.TypedSearchAttributes") +@patch("orchestrator.activities.temporal_manager.SearchAttributePair") +async def test_create_schedule( + mock_search_attribute_pair, + mock_typed_search_attributes, + mock_schedule_spec, + mock_schedule_interval_spec, + mock_schedule_action_start_workflow, + mock_schedule, + mock_parse_frequency, + temporal_manager): + + input_data = { + "schedules": { + "test-schedule": { + "model_id": 1, + "model_name": "test-model-name", + "workflow_type": "test-workflow", + "frequency": "1m", + "data": {"test": "test"} + }, + "test-schedule-invalid-frequency": { + "model_id": 2, + "model_name": "test-model-name", + "workflow_type": "test-workflow", + "frequency": "10y", + "data": {"test": "test"} + } + } + } + + temporal_manager.temporal_client.create_schedule = AsyncMock() + + report = await temporal_manager.create_schedules(input_data) + + temporal_manager.temporal_client.create_schedule.assert_called_once_with( + "test-schedule", + mock_schedule.return_value, + search_attributes=mock_typed_search_attributes.return_value + ) + + mock_schedule.assert_called_once_with( + action=mock_schedule_action_start_workflow.return_value, + spec=mock_schedule_spec.return_value + ) + + mock_schedule_action_start_workflow.assert_has_calls([ + call( + workflow="test-workflow", + args=input_data['schedules']['test-schedule'], + id="test-schedule", + task_queue="test-workflow-queue" + ), + call( + workflow="test-workflow", + args=input_data['schedules']['test-schedule-invalid-frequency'], + id="test-schedule-invalid-frequency", + task_queue="test-workflow-queue" + ) + ]) + + mock_schedule_spec.assert_called_once_with( + intervals=[ + mock_schedule_interval_spec.return_value + ] + ) + + mock_schedule_interval_spec.assert_called_once_with( + every=timedelta(seconds=60) + ) + + mock_parse_frequency.assert_has_calls([ + call("1m"), + call("10y") + ]) + + mock_typed_search_attributes.assert_has_calls([ + call([ + mock_search_attribute_pair.return_value, + mock_search_attribute_pair.return_value, + mock_search_attribute_pair.return_value + ]), + call([ + mock_search_attribute_pair.return_value, + mock_search_attribute_pair.return_value, + mock_search_attribute_pair.return_value + ]) + ]) + + mock_search_attribute_pair.assert_has_calls([ + call( + key=temporal_manager.model_id_id_key, + value=1 + ), + call( + key=temporal_manager.model_name_id_key, + value="test-model-name" + ), + call( + key=temporal_manager.orchestrated_id_key, + value="true" + ) + ]) + + assert report == { + "test-schedule": { + "success": True, + "message": "Schedule created successfully" + }, + "test-schedule-invalid-frequency": { + "success": False, + "message": "Invalid frequency" + } + } + + +@mark.asyncio +@patch("orchestrator.activities.temporal_manager.parse_frequency", + side_effect=parse_frequency) +@patch("orchestrator.activities.temporal_manager.ScheduleIntervalSpec") +async def test_update_schedules( + _mock_schedule_interval_spec, + _mock_parse_frequency, + temporal_manager): + input_mock = MagicMock( + args=MagicMock() + ) + temporal_manager.schedule_handles = { + "test-schedule": MagicMock( + update=AsyncMock( + update=AsyncMock( + side_effect=lambda f: f(input_mock) + ) + ) + ) + } + input_data = { + "schedules": { + "test-schedule": { + "frequency": "1m", + "data": {"test": "test"} + }, + "test-schedule_no_handler": { + "frequency": "1m", + "data": {"test": "test"} + } + } + } + + report = await temporal_manager.update_schedules(input_data) + + temporal_manager.schedule_handles['test-schedule'].update.assert_called_once() + + assert report == { + "test-schedule": { + "success": True, + "message": "Schedule updated successfully" + }, + "test-schedule_no_handler": { + "success": False, + "message": "Schedule test-schedule_no_handler not found" + } + } + + +@mark.asyncio +async def test_delete_schedules(temporal_manager): + temporal_manager.schedule_handles = { + "test-schedule": MagicMock( + delete=AsyncMock() + ) + } + input_data = { + "schedules": [ + "test-schedule", "test-schedule_no_handler" + ] + } + + report = await temporal_manager.delete_schedules(input_data) + + assert report == { + "test-schedule": { + "success": True, + "message": "Schedule deleted successfully" + }, + "test-schedule_no_handler": { + "success": False, + "message": "Schedule test-schedule_no_handler not found" } } diff --git a/tests/orchestrator/utils/test_orchestrator_functions.py b/tests/orchestrator/utils/test_orchestrator_functions.py new file mode 100644 index 0000000..5b27250 --- /dev/null +++ b/tests/orchestrator/utils/test_orchestrator_functions.py @@ -0,0 +1,325 @@ +from unittest.mock import patch, call +from orchestrator.utils.orchestrator_functions import ( + common_config, + scouter, + predictions_batch, + overlap_filter_config, + process_path_priority, + gather_read_tags, + build_tag_config +) + + +def test_common_config(): + config = { + "schedule_name": "test_schedule", + "model_id": "test_model_id", + "model_name": "test_model_name" + } + result = common_config(config) + expected = { + "workflow_type": "scouter", + "schedule_name": "test_schedule", + "frequency": "1m", + "max_retry_policy": 1, + "model_id": "test_model_id", + "model_name": "test_model_name" + } + assert result == expected + + +def test_scouter(): + config = { + "schedule_name": "test_schedule", + "model_id": "test_model_id", + "model_name": "test_model_name", + "filters": [ + { + "filter_name": "test_filter_name", + "policy": "test_policy" + } + ], + "read_tags": [ + { + "tag_name": "test_tag_name", + "aggr_func": "test_aggr_func", + "data_range": [1, 2] + } + ], + "tag_retention_minutes": 10 + } + result = scouter(config) + expected = { + "workflow_type": "scouter", + "schedule_name": "test_schedule", + "frequency": "1m", + "max_retry_policy": 1, + "model_id": "test_model_id", + "model_name": "test_model_name", + "topic": "raw_test_schedule", + "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": { + "aggr_func": "test_aggr_func", + "data_range": [1, 2] + } + } + } + assert result == expected + + +def test_overlap_filter_config(): + config = [ + { + "filter_name": "test_filter_name", + "policy": "test_policy" + }, + { + "filter_name": "test_filter_name2", + "policy": "test_policy2" + } + ] + result = overlap_filter_config({ + "test_filter_name": { + "policy": "test_policy" + } + }, config) + expected = { + "test_filter_name": { + "policy": "test_policy", + "config": {} + }, + "test_filter_name2": { + "policy": "test_policy2", + "config": {} + } + } + assert result == expected + + +def test_process_path_priority(): + config = ["OTHER", "STOP", "CONTINUE"] + result = process_path_priority(config) + expected = ["STOP", "CONTINUE", "REPEAT"] + assert result == expected + + +@patch('orchestrator.utils.orchestrator_functions.overlap_filter_config', + return_value={ + "test_filter_name": { + "policy": "test_policy", + "config": {} + } + }) +@patch('orchestrator.utils.orchestrator_functions.process_path_priority', + return_value=["STOP", "CONTINUE", "REPEAT"]) +def test_predictions_batch(mock_process_path_priority, + mock_overlap_filter_config): + config = { + "schedule_name": "test_schedule", + "workflow_type": "predictions_batch", + "model_id": "test_model_id", + "model_name": "test_model_name", + "query": "test_query", + "write_tags": [ + { + "server_name": "test_server_name", + "type": "prediction", + "addr": "test_addr" + }, + { + "server_name": "test_server_name", + "type": "confidence", + "addr": "test_addr" + } + ], + "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" + } + ], + "path_priority": ["STOP", "CONTINUE", "REPEAT"], + } + + result = predictions_batch(config) + + mock_overlap_filter_config.assert_has_calls([ + 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']) + ]) + mock_overlap_filter_config.assert_has_calls([ + call({ + "API_ERROR": { + "policy": "STOP", + "config": {} + } + }, config['mlflow_predict_filters']) + ]) + mock_process_path_priority.assert_called_once_with(config['path_priority']) + + expected = { + "workflow_type": "predictions_batch", + "schedule_name": "test_schedule", + "frequency": "1m", + "max_retry_policy": 1, + "model_id": "test_model_id", + "model_name": "test_model_name", + "query": "test_query", + "schema": "sientia_data", + "table_name": "predictions", + "retention_time": 60 * 60, + "opc_output_config": { + "test_server_name": { + "prediction_tags": { + "test_addr": { + "data_type": "float" + } + }, + "confidence_tags": { + "test_addr": { + "data_type": "float" + } + } + } + }, + "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": {} + } + }, + "path_priority": ["STOP", "CONTINUE", "REPEAT"] + } + assert result == expected + + +def test_gather_read_tags(): + pipelines = [ + { + "schedule_name": "test_schedule", + "read_tags": [ + { + "server_name": "test_server_name", + "tag_address": "test_tag_address" + } + ] + }, + { + "schedule_name": "test_schedule2", + "read_tags": [ + { + "server_name": "test_server_name2", + "tag_address": "test_tag_address2" + }, + { + "server_name": "test_server_name2", + "tag_address": "test_tag_address3" + } + ] + } + ] + + result = gather_read_tags(pipelines) + + expected = { + "test_server_name:test_tag_address": { + "server_name": "test_server_name", + "tag_address": "test_tag_address", + "topics": ["raw_test_schedule"] + }, + "test_server_name2:test_tag_address2": { + "server_name": "test_server_name2", + "tag_address": "test_tag_address2", + "topics": ["raw_test_schedule2"] + }, + "test_server_name2:test_tag_address3": { + "server_name": "test_server_name2", + "tag_address": "test_tag_address3", + "topics": ["raw_test_schedule2"] + } + } + + assert result == expected + + +def test_build_tag_config(): + tag = { + "server_name": "test_server_name", + "tag_address": "test_tag_address" + } + opc_servers = { + "test_server_name": { + "url": "test_url", + "uri": "test_uri", + "security_spec": { + "test_name": "test_spec" + } + } + } + slot_config = { + "1": {} + } + i = 1 + result = build_tag_config(tag, slot_config, opc_servers, i) + expected = { + "1": { + "test_server_name": { + "name": "test_server_name", + "url": "test_url", + "server_uri": "test_uri", + "tags": { + "test_tag_address": { + "server_name": "test_server_name", + "tag_address": "test_tag_address" + } + }, + "test_name": "test_spec" + } + } + } + assert result == expected diff --git a/tests/orchestrator/workflows/test_orchestrator.py b/tests/orchestrator/workflows/test_orchestrator.py index a9d80f8..0b1d4f8 100644 --- a/tests/orchestrator/workflows/test_orchestrator.py +++ b/tests/orchestrator/workflows/test_orchestrator.py @@ -79,3 +79,111 @@ async def test_run(workflow_mock, orchestrator): start_to_close_timeout=ANY ) ]) + + workflow_mock.execute_local_activity_method.assert_has_calls([ + call( + Activities.process_schedules, + { + 'pipelines': workflow_mock.execute_local_activity_method.return_value + }, + retry_policy=ANY, + start_to_close_timeout=ANY + ) + ]) + + workflow_mock.execute_local_activity_method.assert_has_calls([ + call( + Activities.process_slots, + { + 'opc_servers': workflow_mock.execute_local_activity_method.return_value, + 'active_ingestors': workflow_mock.execute_local_activity_method.return_value, + 'pipelines': workflow_mock.execute_local_activity_method.return_value, + }, + retry_policy=ANY, + start_to_close_timeout=ANY + ) + ]) + + workflow_mock.execute_local_activity_method.assert_has_calls([ + call( + Activities.create_schedule_config, + { + 'current_schedule_config': workflow_mock.execute_local_activity_method.return_value, + 'schedule_config': workflow_mock.execute_local_activity_method.return_value + }, + retry_policy=ANY, + start_to_close_timeout=ANY + ) + ]) + + workflow_mock.execute_local_activity_method.assert_has_calls([ + call( + Activities.create_slot_config, + { + 'current_slot_config': workflow_mock.execute_local_activity_method.return_value, + 'slot_config': workflow_mock.execute_local_activity_method.return_value + }, + retry_policy=ANY, + start_to_close_timeout=ANY + ) + ]) + + workflow_mock.execute_activity_method.assert_has_calls([ + call( + Activities.delete_slots, + { + 'to_delete': + workflow_mock.execute_local_activity_method.return_value['to_delete'] + }, + retry_policy=ANY, + start_to_close_timeout=ANY + ) + ]) + + workflow_mock.execute_activity_method.assert_has_calls([ + call( + Activities.update_slots, + { + 'to_insert': + workflow_mock.execute_local_activity_method.return_value['to_insert'] + }, + retry_policy=ANY, + start_to_close_timeout=ANY + ) + ]) + + workflow_mock.execute_activity_method.assert_has_calls([ + call( + Activities.delete_schedules, + { + 'schedules': + workflow_mock.execute_local_activity_method.return_value['to_delete'] + }, + retry_policy=ANY, + start_to_close_timeout=ANY + ) + ]) + + workflow_mock.execute_activity_method.assert_has_calls([ + call( + Activities.create_schedules, + { + 'schedules': + workflow_mock.execute_local_activity_method.return_value['to_create'] + }, + retry_policy=ANY, + start_to_close_timeout=ANY + ) + ]) + + workflow_mock.execute_activity_method.assert_has_calls([ + call( + Activities.update_schedules, + { + 'schedules': + workflow_mock.execute_local_activity_method.return_value['to_update'] + }, + retry_policy=ANY, + start_to_close_timeout=ANY + ) + ])