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

SIENTIAPDE-1148

Browse Source 
refactor: update temporal configuration and refactor Activities class to support multiple namespaces for improved orchestration
This commit is contained in:
vitor-aignosi
2025-07-11 10:34:40 -03:00
parent d84b721798
commit 7f9efa579a
6 changed files with 247 additions and 162 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
orchestrator/activities/activities.py
View File

@@ -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)

67
orchestrator/activities/formatters.py
View File

@@ -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,

313
orchestrator/activities/temporal_manager.py
View File

@@ -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))

9
orchestrator/utils/connectors_config.py
View File

@@ -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')
}

6
orchestrator/worker/worker.py
View File

@@ -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 = [

6
values.yaml
View File

@@ -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