SIENTIAPDE-1646
SIENTIAPDE-1646 Add support for runtime workflows in TemporalManager by introducing RUNTIME_WORKFLOWS constant and updating schedule creation logic to handle runtime names appropriately.
This commit is contained in:
@@ -25,6 +25,8 @@ with workflow.unsafe.imports_passed_through():
|
||||
from orchestrator.utils.converters import parse_frequency
|
||||
|
||||
|
||||
RUNTIME_WORKFLOWS = ['predictions_batch']
|
||||
|
||||
class TemporalManager(SientiaMonitoring):
|
||||
"""
|
||||
Temporal workflow and schedule management activity.
|
||||
@@ -214,6 +216,8 @@ class TemporalManager(SientiaMonitoring):
|
||||
f'{json.dumps(schedule, indent=4, sort_keys=True)}', metadata=metadata
|
||||
)
|
||||
|
||||
runtime_name = schedule.get('runtime', 'legacy') if workflow_type in RUNTIME_WORKFLOWS else None
|
||||
|
||||
await client.create_schedule(
|
||||
schedule_name,
|
||||
Schedule(
|
||||
@@ -222,7 +226,7 @@ class TemporalManager(SientiaMonitoring):
|
||||
schedule,
|
||||
id=schedule_name,
|
||||
task_queue=build_queue_name(
|
||||
workflow_type, schedule.get('runtime', 'legacy')
|
||||
workflow_type, runtime_name
|
||||
),
|
||||
execution_timeout=timedelta(seconds=execution_timeout_seconds),
|
||||
run_timeout=timedelta(seconds=execution_timeout_seconds),
|
||||
|
||||
Reference in New Issue
Block a user