gitea_admin/sientia-dataops-orchestrator_temporal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

SIENTIAPDE-1150

Browse Source 
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
This commit is contained in:
vitor-aignosi
2025-07-14 13:42:10 -03:00
parent f3497beda0
commit b3069c8a58
6 changed files with 52 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
orchestrator/activities/slot_manager.py
View File

@@ -32,7 +32,7 @@ class SlotManager(Redis):
slot_keys = self.redis_client.keys("slot:opc_tags:*") 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 slot_keys:
if isinstance(slot_keys[0], bytes): 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.info(f"Loaded {len(opc_slots)} OPC slots")
self.logger.debug("Loaded: \n %s", self.logger.debug(
json.dumps(opc_slots, indent=4, sort_keys=True)) f"Loaded: \n {json.dumps(opc_slots, indent=4, sort_keys=True)}")
return opc_slots return opc_slots
@@ -65,7 +65,7 @@ class SlotManager(Redis):
self.logger.info(f"Loaded {len(active_ingestors)} active ingestors") 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 = [] ingestors = []
@@ -105,8 +105,7 @@ class SlotManager(Redis):
"message": "Slot updated successfully" "message": "Slot updated successfully"
} }
except Exception as e: except Exception as e:
self.logger.error("Failed to update slot %s: %s", self.logger.error(f"Failed to update slot {slot}: {str(e)}")
slot, str(e))
report[slot] = { report[slot] = {
"success": False, "success": False,
"message": str(e) "message": str(e)
@@ -114,8 +113,8 @@ class SlotManager(Redis):
self.logger.info(f"Updated {len(to_insert)} OPC slots") self.logger.info(f"Updated {len(to_insert)} OPC slots")
self.logger.debug("Report: \n %s", self.logger.debug(
json.dumps(report, indent=4, sort_keys=True)) f"Report: \n {json.dumps(report, indent=4, sort_keys=True)}")
return report return report
@@ -146,8 +145,7 @@ class SlotManager(Redis):
"message": "Slot deleted successfully" "message": "Slot deleted successfully"
} }
except Exception as e: except Exception as e:
self.logger.error("Failed to delete slot %s: %s", self.logger.error(f"Failed to delete slot {slot}: {str(e)}")
slot, str(e))
report[slot] = { report[slot] = {
"success": False, "success": False,
"message": str(e) "message": str(e)
@@ -155,7 +153,7 @@ class SlotManager(Redis):
self.logger.info(f"Deleted {len(to_delete)} OPC slots") self.logger.info(f"Deleted {len(to_delete)} OPC slots")
self.logger.debug("Report: \n %s", self.logger.debug(
json.dumps(report, indent=4, sort_keys=True)) f"Report: \n {json.dumps(report, indent=4, sort_keys=True)}")
return report return report

59
orchestrator/activities/temporal_manager.py
View File

@@ -80,14 +80,14 @@ class TemporalManager(BaseActivity):
orchestrated_schedules[namespace] = {} orchestrated_schedules[namespace] = {}
self.logger.info( self.logger.info(
"Getting orchestrated schedules for %s", namespace) f"Getting orchestrated schedules for {namespace}")
async for schedule in await client.list_schedules(): async for schedule in await client.list_schedules():
search_attrs = getattr(schedule, "search_attributes", {}) search_attrs = getattr(schedule, "search_attributes", {})
if search_attrs.get("orchestrated", ["false"]) == ["true"]: if search_attrs.get("orchestrated", ["false"]) == ["true"]:
schedule_id = schedule.id 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( handle = client.get_schedule_handle(
schedule_id) schedule_id)
@@ -117,15 +117,14 @@ class TemporalManager(BaseActivity):
await sleep(0.1) await sleep(0.1)
self.logger.info("Found %d orchestrated schedules", self.logger.info(
len(orchestrated_schedules[self.scouter_namespace]) + f"Found {len(orchestrated_schedules[self.scouter_namespace]) + len(orchestrated_schedules[self.laborious_namespace])} orchestrated schedules")
len(orchestrated_schedules[self.laborious_namespace]))
self.logger.debug("Orchestrated schedules: %s", self.logger.debug(
orchestrated_schedules) f"Orchestrated schedules: {orchestrated_schedules}")
self.logger.debug("Schedule handles: %s", self.logger.debug(
self.schedule_handles) f"Schedule handles: {self.schedule_handles}")
return orchestrated_schedules return orchestrated_schedules
@@ -144,7 +143,7 @@ class TemporalManager(BaseActivity):
schedules = orchestrated_schedules.get(namespace, {}) schedules = orchestrated_schedules.get(namespace, {})
self.logger.info( self.logger.info(
"Getting orchestrated schedules for %s", namespace) f"Getting orchestrated schedules for {namespace}")
async for schedule in await client.list_schedules(): async for schedule in await client.list_schedules():
search_attrs = getattr(schedule, "search_attributes", {}) search_attrs = getattr(schedule, "search_attributes", {})
@@ -153,7 +152,7 @@ class TemporalManager(BaseActivity):
if schedule_id not in schedules: if schedule_id not in schedules:
self.logger.info( 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( handle = client.get_schedule_handle(
schedule_id) schedule_id)
@@ -161,7 +160,7 @@ class TemporalManager(BaseActivity):
await handle.delete() await handle.delete()
@activity.defn(name="create_schedules") @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 Create schedules in Temporal
@@ -204,8 +203,8 @@ class TemporalManager(BaseActivity):
try: try:
self.logger.debug(f"Creating schedule {schedule_name}:") self.logger.debug(f"Creating schedule {schedule_name}:")
self.logger.debug(json.dumps( self.logger.debug(
schedule, indent=4, sort_keys=True)) f"{json.dumps(schedule, indent=4, sort_keys=True)}")
await client.create_schedule( await client.create_schedule(
schedule_name, schedule_name,
@@ -236,8 +235,8 @@ class TemporalManager(BaseActivity):
"message": "Schedule created successfully" "message": "Schedule created successfully"
}) })
except Exception as e: except Exception as e:
self.logger.error("Failed to create schedule %s: %s", self.logger.error(
schedule_name, str(e)) f"Failed to create schedule {schedule_name}: {str(e)}")
report.append({ report.append({
"namespace": namespace, "namespace": namespace,
"schedule_name": schedule_name, "schedule_name": schedule_name,
@@ -247,13 +246,13 @@ class TemporalManager(BaseActivity):
self.logger.info(f"Processed {len(schedules_to_create)} schedules") self.logger.info(f"Processed {len(schedules_to_create)} schedules")
self.logger.debug("\n %s", self.logger.debug(
json.dumps(report, indent=4, sort_keys=True)) f"\n {json.dumps(report, indent=4, sort_keys=True)}")
return report return report
@activity.defn(name="update_schedules") @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 Update schedules in Temporal
@@ -293,8 +292,8 @@ class TemporalManager(BaseActivity):
if hasattr(schedule_action, "args"): if hasattr(schedule_action, "args"):
self.logger.debug("New schedule:") self.logger.debug("New schedule:")
self.logger.debug(json.dumps( self.logger.debug(
schedule, indent=4, sort_keys=True)) # NOSONAR f"{json.dumps(schedule, indent=4, sort_keys=True)}") # NOSONAR
schedule_action.args = [schedule] schedule_action.args = [schedule]
@@ -319,8 +318,8 @@ class TemporalManager(BaseActivity):
"message": "Schedule updated successfully" "message": "Schedule updated successfully"
}) })
except Exception as e: except Exception as e:
self.logger.error("Failed to update schedule %s: %s", self.logger.error(
schedule_name, str(e)) f"Failed to update schedule {schedule_name}: {str(e)}")
report.append({ report.append({
"namespace": namespace, "namespace": namespace,
"schedule_name": schedule_name, "schedule_name": schedule_name,
@@ -330,13 +329,13 @@ class TemporalManager(BaseActivity):
self.logger.info(f"Processed {len(schedules_to_update)} schedules") self.logger.info(f"Processed {len(schedules_to_update)} schedules")
self.logger.debug("\n %s", self.logger.debug(
json.dumps(report, indent=4, sort_keys=True)) f"\n {json.dumps(report, indent=4, sort_keys=True)}")
return report return report
@activity.defn(name="delete_schedules") @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 Delete schedules in Temporal
@@ -379,8 +378,8 @@ class TemporalManager(BaseActivity):
"message": "Schedule deleted successfully" "message": "Schedule deleted successfully"
}) })
except Exception as e: except Exception as e:
self.logger.error("Failed to delete schedule %s: %s", self.logger.error(
schedule_name, str(e)) f"Failed to delete schedule {schedule_name}: {str(e)}")
report.append({ report.append({
"namespace": namespace, "namespace": namespace,
"schedule_name": schedule_name, "schedule_name": schedule_name,
@@ -390,7 +389,7 @@ class TemporalManager(BaseActivity):
self.logger.info(f"Processed {len(schedules_to_delete)} schedules") self.logger.info(f"Processed {len(schedules_to_delete)} schedules")
self.logger.debug("\n %s", self.logger.debug(
json.dumps(report, indent=4, sort_keys=True)) f"\n {json.dumps(report, indent=4, sort_keys=True)}")
return report return report

10
orchestrator/utils/connectors_config.py
View File

@@ -5,15 +5,15 @@ def build_redis_config():
return { return {
'host': getenv('REDIS_HOST', 'localhost'), 'host': getenv('REDIS_HOST', 'localhost'),
'port': int(getenv('REDIS_PORT', '6379')), 'port': int(getenv('REDIS_PORT', '6379')),
'username': getenv('REDIS_USERNAME', None), 'username': getenv('REDIS_USERNAME', 'default'),
'password': getenv('REDIS_PASSWORD', None) 'password': getenv('REDIS_PASSWORD', 'bdnZOpcyiL')
} }
def build_mongodb_config(): def build_mongodb_config():
username = getenv('MONGODB_USERNAME', 'sientia') username = getenv('MONGODB_USERNAME', 'root')
password = getenv('MONGODB_PASSWORD', 'sientia') password = getenv('MONGODB_PASSWORD', 'wKZDbMNU1c')
uri = getenv('MONGODB_URL', 'localhost:27017') uri = getenv('MONGODB_URL', 'localhost:27018')
connection_string = f'mongodb://{username}:{password}@{uri}' connection_string = f'mongodb://{username}:{password}@{uri}'
return { return {

7
orchestrator/worker/worker.py
View File

@@ -69,11 +69,15 @@ async def main():
# MongoDB # MongoDB
activities.aggregate_documents_in_mongodb, activities.aggregate_documents_in_mongodb,
activities.find_documents_in_mongodb, activities.find_documents_in_mongodb,
activities.update_pipelines_timestamps,
activities.create_pipelines_timestamps,
activities.delete_pipelines_timestamps,
# Temporal # Temporal
activities.load_schedule, activities.load_schedule,
activities.create_schedules, activities.create_schedules,
activities.update_schedules, activities.update_schedules,
activities.delete_schedules, activities.delete_schedules,
activities.normalize_schedules,
# Formatters # Formatters
activities.process_schedules, activities.process_schedules,
activities.process_slots, activities.process_slots,
@@ -81,6 +85,7 @@ async def main():
activities.create_slot_config, activities.create_slot_config,
activities.report_schedule_orchestration, activities.report_schedule_orchestration,
activities.report_slot_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. # If an exception occurs in any of the worker handlers, it will be propagated here.
await asyncio.gather(*handlers) await asyncio.gather(*handlers)
except BaseException as e: 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: finally:
if notification_handler: if notification_handler:
notification_handler.shutdown() notification_handler.shutdown()

2
tests/orchestrator/activities/test_couchbase.py
View File

@@ -25,7 +25,7 @@ def test_shutdown_failure(couchbase):
couchbase.cluster.close.side_effect = Exception("Test error") couchbase.cluster.close.side_effect = Exception("Test error")
couchbase.shutdown() couchbase.shutdown()
couchbase.logger.error.assert_called_once_with( 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 @mark.asyncio

2
values.yaml
View File

@@ -11,7 +11,7 @@ image:
# This sets the pull policy for images. # This sets the pull policy for images.
pullPolicy: Always pullPolicy: Always
# Overrides the image tag whose default is the chart appVersion. # 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/ # 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: imagePullSecrets: