From 7f9efa579a171a2c306e16ea39628d613dc80a81 Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Fri, 11 Jul 2025 10:34:40 -0300 Subject: [PATCH] SIENTIAPDE-1148 refactor: update temporal configuration and refactor Activities class to support multiple namespaces for improved orchestration --- orchestrator/activities/activities.py | 8 +- orchestrator/activities/formatters.py | 67 +++-- orchestrator/activities/temporal_manager.py | 313 +++++++++++--------- orchestrator/utils/connectors_config.py | 9 + orchestrator/worker/worker.py | 6 +- values.yaml | 6 +- 6 files changed, 247 insertions(+), 162 deletions(-) diff --git a/orchestrator/activities/activities.py b/orchestrator/activities/activities.py index 5049e3f..5b0ded5 100644 --- a/orchestrator/activities/activities.py +++ b/orchestrator/activities/activities.py @@ -17,7 +17,7 @@ class Activities( # Couchbase, TemporalManager, SlotManager, Formatters, MongoDB): def __init__(self, - temporal_client: Client, + temporal_config: dict[str, Any], # couchbase_config: dict[str, Any], redis_config: dict[str, Any], mongodb_config: dict[str, Any], @@ -32,7 +32,9 @@ class Activities( # Couchbase, # notification_handler=notification_handler) TemporalManager.__init__(self, - temporal_client=temporal_client, + host=temporal_config['host'], + scouter_namespace=temporal_config['scouter_namespace'], + laborious_namespace=temporal_config['laborious_namespace'], logger=logger, notification_handler=notification_handler) @@ -45,6 +47,8 @@ class Activities( # Couchbase, notification_handler=notification_handler) Formatters.__init__(self, + scouter_namespace=temporal_config['scouter_namespace'], + laborious_namespace=temporal_config['laborious_namespace'], logger=logger, notification_handler=notification_handler) diff --git a/orchestrator/activities/formatters.py b/orchestrator/activities/formatters.py index 8a06865..da6474b 100644 --- a/orchestrator/activities/formatters.py +++ b/orchestrator/activities/formatters.py @@ -14,7 +14,12 @@ with workflow.unsafe.imports_passed_through(): class Formatters(BaseActivity): - def __init__(self, logger: Logger, notification_handler: NotificationHandler): + def __init__(self, + scouter_namespace: str, + laborious_namespace: str, + logger: Logger, notification_handler: NotificationHandler): + self.scouter_namespace = scouter_namespace + self.laborious_namespace = laborious_namespace BaseActivity.__init__(self, logger=logger, notification_handler=notification_handler) @@ -37,14 +42,18 @@ class Formatters(BaseActivity): pipelines = input_data['pipelines'] - schedule_config = {} + schedule_config = { + self.scouter_namespace: {}, + self.laborious_namespace: {} + } for pipeline in pipelines: if pipeline['workflow_type'] == 'scouter': - schedule_config[pipeline['schedule_name']] = scouter(pipeline) + schedule_config[self.scouter_namespace][pipeline['schedule_name']] = scouter( + pipeline) elif pipeline['workflow_type'] == 'predictions_batch': - schedule_config[pipeline['schedule_name'] - ] = predictions_batch(pipeline) + schedule_config[self.laborious_namespace][pipeline['schedule_name'] + ] = predictions_batch(pipeline) self.logger.info("Processed schedules") @@ -134,30 +143,40 @@ class Formatters(BaseActivity): current_schedule_config = input_data['current_schedule_config'] schedule_config = input_data['schedule_config'] - to_update = {} - to_create = {} - to_delete = [] + to_update = { + self.scouter_namespace: {}, + self.laborious_namespace: {} + } + to_create = { + self.scouter_namespace: {}, + self.laborious_namespace: {} + } + to_delete = { + self.scouter_namespace: [], + self.laborious_namespace: [] + } - for schedule_name, schedule in schedule_config.items(): - if schedule_name in current_schedule_config: - self.logger.debug(f"{current_schedule_config[schedule_name]}") - old_config = current_schedule_config[schedule_name]['data'] + for namespace, schedules in schedule_config.items(): + for schedule_name, schedule in schedules.items(): + if schedule_name in current_schedule_config[namespace]: + old_config = current_schedule_config[namespace][schedule_name]['data'] - self.logger.debug(f"Comparing {schedule_name}:") - self.logger.debug(json.dumps( - old_config, indent=4, sort_keys=True)) - self.logger.debug(json.dumps( - schedule, indent=4, sort_keys=True)) + self.logger.debug(f"Comparing {schedule_name}:") + self.logger.debug(json.dumps( + old_config, indent=4, sort_keys=True)) + self.logger.debug(json.dumps( + schedule, indent=4, sort_keys=True)) - if schedule != old_config: - to_update[schedule_name] = schedule + if schedule != old_config: + to_update[namespace][schedule_name] = schedule - elif schedule_name not in current_schedule_config: - to_create[schedule_name] = schedule + elif schedule_name not in current_schedule_config: + to_create[namespace][schedule_name] = schedule - for schedule_name in current_schedule_config: - if schedule_name not in schedule_config: - to_delete.append(schedule_name) + for namespace, schedules in current_schedule_config.items(): + for schedule_name in schedules: + if schedule_name not in schedule_config[namespace]: + to_delete[namespace].append(schedule_name) output = { "to_update": to_update, diff --git a/orchestrator/activities/temporal_manager.py b/orchestrator/activities/temporal_manager.py index fc4c5a1..73d81d3 100644 --- a/orchestrator/activities/temporal_manager.py +++ b/orchestrator/activities/temporal_manager.py @@ -1,4 +1,5 @@ from asyncio import sleep +from http import client from temporalio import activity, workflow from temporalio.client import ( Client, Schedule, ScheduleActionStartWorkflow, ScheduleIntervalSpec, ScheduleSpec, ScheduleUpdate, ScheduleUpdateInput) @@ -17,12 +18,18 @@ with workflow.unsafe.imports_passed_through(): class TemporalManager(BaseActivity): - def __init__(self, temporal_client: Client, logger: Logger, - notification_handler: NotificationHandler): + def __init__(self, host: str, scouter_namespace: str, laborious_namespace: str, + logger: Logger, notification_handler: NotificationHandler): - self.temporal_client = temporal_client + self.host = host + self.scouter_namespace = scouter_namespace + self.laborious_namespace = laborious_namespace + self.temporal_clients = {} - self.schedule_handles = {} + self.schedule_handles = { + self.scouter_namespace: {}, + self.laborious_namespace: {} + } self.model_id_id_key = SearchAttributeKey.for_keyword("model_id") self.model_name_id_key = SearchAttributeKey.for_keyword("model_name") @@ -33,6 +40,18 @@ class TemporalManager(BaseActivity): logger=logger, notification_handler=notification_handler) + async def connect_to_temporal(self): + self.temporal_clients = { + self.scouter_namespace: await Client.connect( + target_host=self.host, + namespace=self.scouter_namespace + ), + self.laborious_namespace: await Client.connect( + target_host=self.host, + namespace=self.laborious_namespace + ) + } + @activity.defn(name="load_schedule") async def load_schedule(self) -> dict[str, Any]: """ @@ -48,39 +67,46 @@ class TemporalManager(BaseActivity): orchestrated_schedules = {} - async for schedule in await self.temporal_client.list_schedules(): - search_attrs = getattr(schedule, "search_attributes", {}) - if search_attrs.get("orchestrated", ["false"]) == ["true"]: - schedule_id = schedule.id + for namespace, client in self.temporal_clients.items(): + orchestrated_schedules[namespace] = {} - self.logger.debug("Schedule id: %s", schedule_id) + self.logger.info( + "Getting orchestrated schedules for %s", namespace) - handle = self.temporal_client.get_schedule_handle(schedule_id) + async for schedule in await client.list_schedules(): + search_attrs = getattr(schedule, "search_attributes", {}) + if search_attrs.get("orchestrated", ["false"]) == ["true"]: + schedule_id = schedule.id - self.logger.debug("Handle acquired") + self.logger.debug("Schedule id: %s", schedule_id) - self.schedule_handles[schedule_id] = handle + handle = client.get_schedule_handle( + schedule_id) - self.logger.debug("Describing schedule...") + self.logger.debug("Handle acquired") - desc = await handle.describe( - rpc_timeout=timedelta(seconds=60) - ) + self.schedule_handles[namespace][schedule_id] = handle - self.logger.debug("Parsing args...") + self.logger.debug("Describing schedule...") - for arg in desc.schedule.action.args: - data = MessageToDict(arg)['data'] - data = base64.b64decode(data).decode('utf-8') + desc = await handle.describe( + rpc_timeout=timedelta(seconds=60) + ) - frequency = desc.schedule.spec.intervals[0].every.seconds + self.logger.debug("Parsing args...") - orchestrated_schedules[schedule_id] = { - 'frequency': frequency, - 'data': json.loads(data), - } + for arg in desc.schedule.action.args: + data = MessageToDict(arg)['data'] + data = base64.b64decode(data).decode('utf-8') - await sleep(0.1) + frequency = desc.schedule.spec.intervals[0].every.seconds + + orchestrated_schedules[schedule_id] = { + 'frequency': frequency, + 'data': json.loads(data), + } + + await sleep(0.1) self.logger.info("Found %d orchestrated schedules", len(orchestrated_schedules)) @@ -104,66 +130,73 @@ class TemporalManager(BaseActivity): - dict[str, Any]: A report of the created schedules. """ - schedules = input_data['schedules'] + schedules_to_create = 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'] + for namespace, schedules in schedules_to_create.items(): + client = self.temporal_clients.get(namespace) - try: - self.logger.debug(f"Creating schedule {schedule_name}:") - self.logger.debug(json.dumps( - schedule, indent=4, sort_keys=True)) - await self.temporal_client.create_schedule( - schedule_name, - Schedule( - action=ScheduleActionStartWorkflow( - workflow_type, - schedule, - id=schedule_name, - task_queue=f"{workflow_type}-queue", - execution_timeout=timedelta(minutes=2) - ), - spec=ScheduleSpec( - intervals=[ - ScheduleIntervalSpec( - every=timedelta(seconds=parse_frequency( - schedule.get('frequency', '1m'))) - ) - ] - ) + if not client: + raise ValueError(f"Temporal client for {namespace} not found") + + for schedule_name, schedule in schedules.items(): + search_attributes = TypedSearchAttributes([ + SearchAttributePair( + key=self.model_id_id_key, + value=schedule['model_id'] ), - search_attributes=search_attributes - ) + SearchAttributePair( + key=self.model_name_id_key, + value=schedule['model_name'] + ), + SearchAttributePair( + key=self.orchestrated_id_key, + value="true" + ) + ]) + workflow_type = schedule['workflow_type'] - 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) - } + try: + self.logger.debug(f"Creating schedule {schedule_name}:") + self.logger.debug(json.dumps( + schedule, indent=4, sort_keys=True)) - self.logger.info(f"Processed {len(schedules)} schedules") + await client.create_schedule( + schedule_name, + Schedule( + action=ScheduleActionStartWorkflow( + workflow_type, + schedule, + id=schedule_name, + task_queue=f"{workflow_type}-queue", + execution_timeout=timedelta(minutes=2) + ), + 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_to_create)} schedules") self.logger.debug("\n %s", json.dumps(report, indent=4, sort_keys=True)) @@ -184,55 +217,61 @@ class TemporalManager(BaseActivity): - dict[str, Any]: A report of the updated schedules. """ - schedules = input_data['schedules'] + schedules_to_update = input_data['schedules'] report = {} - for schedule_name, schedule in schedules.items(): - try: - handler = self.schedule_handles.get(schedule_name) + for namespace, schedules in schedules_to_update.items(): + schedule_handles = self.schedule_handles.get(namespace) - if not handler: - raise ValueError(f"Schedule {schedule_name} not found") + if not schedule_handles: + raise ValueError(f"Schedule handles for {namespace} not found") - async def update_schedule(input_data: ScheduleUpdateInput) -> ScheduleUpdate: - schedule_action = input_data.description.schedule.action + for schedule_name, schedule in schedules.items(): + try: + handler = schedule_handles.get(schedule_name) - self.logger.debug("Updating schedule:") + if not handler: + raise ValueError(f"Schedule {schedule_name} not found") - if hasattr(schedule_action, "args"): - self.logger.debug("New schedule:") - self.logger.debug(json.dumps( - schedule, indent=4, sort_keys=True)) + async def update_schedule(input_data: ScheduleUpdateInput) -> ScheduleUpdate: + schedule_action = input_data.description.schedule.action - schedule_action.args = [schedule] + self.logger.debug("Updating schedule:") - input_data.description.schedule.spec.intervals = [ - ScheduleIntervalSpec( - every=timedelta(seconds=parse_frequency( - schedule.get('frequency', '1m'))) - ) - ] + if hasattr(schedule_action, "args"): + self.logger.debug("New schedule:") + self.logger.debug(json.dumps( + schedule, indent=4, sort_keys=True)) - return ScheduleUpdate(schedule=input_data.description.schedule) + schedule_action.args = [schedule] - await handler.update(update_schedule) + input_data.description.schedule.spec.intervals = [ + ScheduleIntervalSpec( + every=timedelta(seconds=parse_frequency( + schedule.get('frequency', '1m'))) + ) + ] - del update_schedule + return ScheduleUpdate(schedule=input_data.description.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) - } + await handler.update(update_schedule) - self.logger.info(f"Processed {len(schedules)} schedules") + 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_to_update)} schedules") self.logger.debug("\n %s", json.dumps(report, indent=4, sort_keys=True)) @@ -253,34 +292,40 @@ class TemporalManager(BaseActivity): - dict[str, Any]: A report of the deleted schedules. """ - schedules = input_data['schedules'] + schedules_to_delete = input_data['schedules'] report = {} - for schedule_name in schedules: - try: - handler = self.schedule_handles.get(schedule_name) + for namespace, schedules in schedules_to_delete.items(): + schedule_handles = self.schedule_handles.get(namespace) - if not handler: - raise ValueError(f"Schedule {schedule_name} not found") + if not schedule_handles: + raise ValueError(f"Schedule handles for {namespace} not found") - await handler.delete() + for schedule_name in schedules: + try: + handler = schedule_handles.get(schedule_name) - del self.schedule_handles[schedule_name] + if not handler: + raise ValueError(f"Schedule {schedule_name} not found") - 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) - } + await handler.delete() - self.logger.info(f"Processed {len(schedules)} schedules") + del self.schedule_handles[namespace][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_to_delete)} schedules") self.logger.debug("\n %s", json.dumps(report, indent=4, sort_keys=True)) diff --git a/orchestrator/utils/connectors_config.py b/orchestrator/utils/connectors_config.py index 3808f72..fe8225a 100644 --- a/orchestrator/utils/connectors_config.py +++ b/orchestrator/utils/connectors_config.py @@ -28,3 +28,12 @@ def build_couchbase_config(): 'username': getenv('COUCHBASE_USERNAME', 'sientia'), 'password': getenv('COUCHBASE_PASSWORD', 'sientia') } + + +def build_temporal_config(): + return { + 'host': getenv('TEMPORAL_HOST', 'localhost:7233'), + 'namespace': getenv('TEMPORAL_NAMESPACE', 'default'), + 'scouter_namespace': getenv('TEMPORAL_SCOUTER_NAMESPACE', 'scouter'), + 'laborious_namespace': getenv('TEMPORAL_LABORIOUS_NAMESPACE', 'laborious') + } diff --git a/orchestrator/worker/worker.py b/orchestrator/worker/worker.py index 8aa019f..f4a2804 100644 --- a/orchestrator/worker/worker.py +++ b/orchestrator/worker/worker.py @@ -1,6 +1,8 @@ from temporalio import workflow, client from temporalio.worker import Worker +from orchestrator.utils.connectors_config import build_temporal_config + with workflow.unsafe.imports_passed_through(): import os import sys @@ -40,13 +42,15 @@ async def main(): logger.info('Starting Activities...') activities = Activities( - temporal_client=temporal_client, + temporal_config=build_temporal_config(), redis_config=build_redis_config(), mongodb_config=build_mongodb_config(), logger=logger, notification_handler=notification_handler ) + await activities.connect_to_temporal() + logger.info('Starting Workers...') workers = [ diff --git a/values.yaml b/values.yaml index 7bb0447..067b888 100644 --- a/values.yaml +++ b/values.yaml @@ -123,7 +123,7 @@ env: - name: GITHUB_REPO_URL value: "git@github.com:Aignosi/sientia-dataops-orchestrator_temporal.git" - name: GITHUB_BRANCH - value: "SIENTIAPDE-1141-criar-testes-de-carga" + value: "SIENTIAPDE-1148-separar-scouter-laborious-e-orchestrator-por-namespaces" - name: PYTHON_APP value: "orchestrator.worker.worker" @@ -171,6 +171,10 @@ env: value: "temporal-frontend.temporal.svc.cluster.local:7233" - name: TEMPORAL_NAMESPACE value: "default" + - name: TEMPORAL_SCOUTER_NAMESPACE + value: "scouter" + - name: TEMPORAL_LABORIOUS_NAMESPACE + value: "laborious" ssh: enabled: true