From b3069c8a588e1781c32c0b8eb022798a09be905e Mon Sep 17 00:00:00 2001 From: vitor-aignosi Date: Mon, 14 Jul 2025 13:42:10 -0300 Subject: [PATCH] SIENTIAPDE-1150 chore: update image tag to 0.2.4 in values.yaml; refactor logging in slot_manager and temporal_manager for improved readability; modify MongoDB and Redis configurations for better defaults; enhance worker.py with new activities and error handling --- orchestrator/activities/slot_manager.py | 22 ++++--- orchestrator/activities/temporal_manager.py | 59 +++++++++---------- orchestrator/utils/connectors_config.py | 10 ++-- orchestrator/worker/worker.py | 7 ++- .../orchestrator/activities/test_couchbase.py | 2 +- values.yaml | 2 +- 6 files changed, 52 insertions(+), 50 deletions(-) diff --git a/orchestrator/activities/slot_manager.py b/orchestrator/activities/slot_manager.py index d8484cb..cf2e49f 100644 --- a/orchestrator/activities/slot_manager.py +++ b/orchestrator/activities/slot_manager.py @@ -32,7 +32,7 @@ class SlotManager(Redis): slot_keys = self.redis_client.keys("slot:opc_tags:*") - self.logger.debug("Slot keys: %s", slot_keys) + self.logger.debug(f"Slot keys: {slot_keys}") if slot_keys: if isinstance(slot_keys[0], bytes): @@ -45,8 +45,8 @@ class SlotManager(Redis): self.logger.info(f"Loaded {len(opc_slots)} OPC slots") - self.logger.debug("Loaded: \n %s", - json.dumps(opc_slots, indent=4, sort_keys=True)) + self.logger.debug( + f"Loaded: \n {json.dumps(opc_slots, indent=4, sort_keys=True)}") return opc_slots @@ -65,7 +65,7 @@ class SlotManager(Redis): self.logger.info(f"Loaded {len(active_ingestors)} active ingestors") - self.logger.debug("Active ingestors: \n %s", active_ingestors) + self.logger.debug(f"Active ingestors: \n {active_ingestors}") ingestors = [] @@ -105,8 +105,7 @@ class SlotManager(Redis): "message": "Slot updated successfully" } except Exception as e: - self.logger.error("Failed to update slot %s: %s", - slot, str(e)) + self.logger.error(f"Failed to update slot {slot}: {str(e)}") report[slot] = { "success": False, "message": str(e) @@ -114,8 +113,8 @@ class SlotManager(Redis): self.logger.info(f"Updated {len(to_insert)} OPC slots") - self.logger.debug("Report: \n %s", - json.dumps(report, indent=4, sort_keys=True)) + self.logger.debug( + f"Report: \n {json.dumps(report, indent=4, sort_keys=True)}") return report @@ -146,8 +145,7 @@ class SlotManager(Redis): "message": "Slot deleted successfully" } except Exception as e: - self.logger.error("Failed to delete slot %s: %s", - slot, str(e)) + self.logger.error(f"Failed to delete slot {slot}: {str(e)}") report[slot] = { "success": False, "message": str(e) @@ -155,7 +153,7 @@ class SlotManager(Redis): self.logger.info(f"Deleted {len(to_delete)} OPC slots") - self.logger.debug("Report: \n %s", - json.dumps(report, indent=4, sort_keys=True)) + self.logger.debug( + f"Report: \n {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 59b6274..7b4b929 100644 --- a/orchestrator/activities/temporal_manager.py +++ b/orchestrator/activities/temporal_manager.py @@ -80,14 +80,14 @@ class TemporalManager(BaseActivity): orchestrated_schedules[namespace] = {} self.logger.info( - "Getting orchestrated schedules for %s", namespace) + f"Getting orchestrated schedules for {namespace}") 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("Schedule id: %s", schedule_id) + self.logger.debug(f"Schedule id: {schedule_id}") handle = client.get_schedule_handle( schedule_id) @@ -117,15 +117,14 @@ class TemporalManager(BaseActivity): await sleep(0.1) - self.logger.info("Found %d orchestrated schedules", - len(orchestrated_schedules[self.scouter_namespace]) + - len(orchestrated_schedules[self.laborious_namespace])) + self.logger.info( + f"Found {len(orchestrated_schedules[self.scouter_namespace]) + len(orchestrated_schedules[self.laborious_namespace])} orchestrated schedules") - self.logger.debug("Orchestrated schedules: %s", - orchestrated_schedules) + self.logger.debug( + f"Orchestrated schedules: {orchestrated_schedules}") - self.logger.debug("Schedule handles: %s", - self.schedule_handles) + self.logger.debug( + f"Schedule handles: {self.schedule_handles}") return orchestrated_schedules @@ -144,7 +143,7 @@ class TemporalManager(BaseActivity): schedules = orchestrated_schedules.get(namespace, {}) self.logger.info( - "Getting orchestrated schedules for %s", namespace) + f"Getting orchestrated schedules for {namespace}") async for schedule in await client.list_schedules(): search_attrs = getattr(schedule, "search_attributes", {}) @@ -153,7 +152,7 @@ class TemporalManager(BaseActivity): if schedule_id not in schedules: self.logger.info( - "Schedule %s not found in mongo db, cleaning up", schedule_id) + f"Schedule {schedule_id} not found in mongo db, cleaning up") handle = client.get_schedule_handle( schedule_id) @@ -161,7 +160,7 @@ class TemporalManager(BaseActivity): await handle.delete() @activity.defn(name="create_schedules") - async def create_schedules(self, input_data: dict[str, Any]) -> dict[str, Any]: + async def create_schedules(self, input_data: dict[str, Any]) -> list[dict[str, Any]]: """ Create schedules in Temporal @@ -204,8 +203,8 @@ class TemporalManager(BaseActivity): try: self.logger.debug(f"Creating schedule {schedule_name}:") - self.logger.debug(json.dumps( - schedule, indent=4, sort_keys=True)) + self.logger.debug( + f"{json.dumps(schedule, indent=4, sort_keys=True)}") await client.create_schedule( schedule_name, @@ -236,8 +235,8 @@ class TemporalManager(BaseActivity): "message": "Schedule created successfully" }) except Exception as e: - self.logger.error("Failed to create schedule %s: %s", - schedule_name, str(e)) + self.logger.error( + f"Failed to create schedule {schedule_name}: {str(e)}") report.append({ "namespace": namespace, "schedule_name": schedule_name, @@ -247,13 +246,13 @@ class TemporalManager(BaseActivity): self.logger.info(f"Processed {len(schedules_to_create)} schedules") - self.logger.debug("\n %s", - json.dumps(report, indent=4, sort_keys=True)) + self.logger.debug( + f"\n {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]: + async def update_schedules(self, input_data: dict[str, Any]) -> list[dict[str, Any]]: """ Update schedules in Temporal @@ -293,8 +292,8 @@ class TemporalManager(BaseActivity): if hasattr(schedule_action, "args"): self.logger.debug("New schedule:") - self.logger.debug(json.dumps( - schedule, indent=4, sort_keys=True)) # NOSONAR + self.logger.debug( + f"{json.dumps(schedule, indent=4, sort_keys=True)}") # NOSONAR schedule_action.args = [schedule] @@ -319,8 +318,8 @@ class TemporalManager(BaseActivity): "message": "Schedule updated successfully" }) except Exception as e: - self.logger.error("Failed to update schedule %s: %s", - schedule_name, str(e)) + self.logger.error( + f"Failed to update schedule {schedule_name}: {str(e)}") report.append({ "namespace": namespace, "schedule_name": schedule_name, @@ -330,13 +329,13 @@ class TemporalManager(BaseActivity): self.logger.info(f"Processed {len(schedules_to_update)} schedules") - self.logger.debug("\n %s", - json.dumps(report, indent=4, sort_keys=True)) + self.logger.debug( + f"\n {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]: + async def delete_schedules(self, input_data: dict[str, Any]) -> list[dict[str, Any]]: """ Delete schedules in Temporal @@ -379,8 +378,8 @@ class TemporalManager(BaseActivity): "message": "Schedule deleted successfully" }) except Exception as e: - self.logger.error("Failed to delete schedule %s: %s", - schedule_name, str(e)) + self.logger.error( + f"Failed to delete schedule {schedule_name}: {str(e)}") report.append({ "namespace": namespace, "schedule_name": schedule_name, @@ -390,7 +389,7 @@ class TemporalManager(BaseActivity): self.logger.info(f"Processed {len(schedules_to_delete)} schedules") - self.logger.debug("\n %s", - json.dumps(report, indent=4, sort_keys=True)) + self.logger.debug( + f"\n {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 3b7551a..103d20b 100644 --- a/orchestrator/utils/connectors_config.py +++ b/orchestrator/utils/connectors_config.py @@ -5,15 +5,15 @@ def build_redis_config(): return { 'host': getenv('REDIS_HOST', 'localhost'), 'port': int(getenv('REDIS_PORT', '6379')), - 'username': getenv('REDIS_USERNAME', None), - 'password': getenv('REDIS_PASSWORD', None) + 'username': getenv('REDIS_USERNAME', 'default'), + 'password': getenv('REDIS_PASSWORD', 'bdnZOpcyiL') } def build_mongodb_config(): - username = getenv('MONGODB_USERNAME', 'sientia') - password = getenv('MONGODB_PASSWORD', 'sientia') - uri = getenv('MONGODB_URL', 'localhost:27017') + username = getenv('MONGODB_USERNAME', 'root') + password = getenv('MONGODB_PASSWORD', 'wKZDbMNU1c') + uri = getenv('MONGODB_URL', 'localhost:27018') connection_string = f'mongodb://{username}:{password}@{uri}' return { diff --git a/orchestrator/worker/worker.py b/orchestrator/worker/worker.py index 76c88f1..a0f58bd 100644 --- a/orchestrator/worker/worker.py +++ b/orchestrator/worker/worker.py @@ -69,11 +69,15 @@ async def main(): # MongoDB activities.aggregate_documents_in_mongodb, activities.find_documents_in_mongodb, + activities.update_pipelines_timestamps, + activities.create_pipelines_timestamps, + activities.delete_pipelines_timestamps, # Temporal activities.load_schedule, activities.create_schedules, activities.update_schedules, activities.delete_schedules, + activities.normalize_schedules, # Formatters activities.process_schedules, activities.process_slots, @@ -81,6 +85,7 @@ async def main(): activities.create_slot_config, activities.report_schedule_orchestration, activities.report_slot_orchestration, + activities.format_schedule_config, ] ) ] @@ -96,7 +101,7 @@ async def main(): # If an exception occurs in any of the worker handlers, it will be propagated here. await asyncio.gather(*handlers) except BaseException as e: - logger.error("An unhandled exception occurred: %s", e, exc_info=True) + logger.error(f"An unhandled exception occurred: {e}", exc_info=True) finally: if notification_handler: notification_handler.shutdown() diff --git a/tests/orchestrator/activities/test_couchbase.py b/tests/orchestrator/activities/test_couchbase.py index 91b418a..7f78ec5 100644 --- a/tests/orchestrator/activities/test_couchbase.py +++ b/tests/orchestrator/activities/test_couchbase.py @@ -25,7 +25,7 @@ def test_shutdown_failure(couchbase): couchbase.cluster.close.side_effect = Exception("Test error") couchbase.shutdown() couchbase.logger.error.assert_called_once_with( - "Failed to close Couchbase connection: %s", couchbase.cluster.close.side_effect) + f"Failed to close Couchbase connection: {couchbase.cluster.close.side_effect}") @mark.asyncio diff --git a/values.yaml b/values.yaml index 067b888..bff3de6 100644 --- a/values.yaml +++ b/values.yaml @@ -11,7 +11,7 @@ image: # This sets the pull policy for images. pullPolicy: Always # Overrides the image tag whose default is the chart appVersion. - tag: "0.2.0" + tag: "0.2.4" # This is for the secrets for pulling an image from a private repository more information can be found here: https://kubernetes.io/docs/tasks/configure-pod-container/pull-image-private-registry/ imagePullSecrets: