diff --git a/ESSArch_Core/WorkflowEngine/migrations/0089_alter_processstep_queue_alter_processtask_queue.py b/ESSArch_Core/WorkflowEngine/migrations/0089_alter_processstep_queue_alter_processtask_queue.py new file mode 100644 index 000000000..a5b4a49f4 --- /dev/null +++ b/ESSArch_Core/WorkflowEngine/migrations/0089_alter_processstep_queue_alter_processtask_queue.py @@ -0,0 +1,23 @@ +# Generated by Django 5.2.14 on 2026-06-01 17:19 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('WorkflowEngine', '0088_remove_processstep_type_processstep_label_and_more'), + ] + + operations = [ + migrations.AlterField( + model_name='processstep', + name='queue', + field=models.CharField(blank=True, default='celery', max_length=255, null=True), + ), + migrations.AlterField( + model_name='processtask', + name='queue', + field=models.CharField(blank=True, default='celery', max_length=255, null=True), + ), + ] diff --git a/ESSArch_Core/WorkflowEngine/models.py b/ESSArch_Core/WorkflowEngine/models.py index 6e4f32551..8519cf264 100644 --- a/ESSArch_Core/WorkflowEngine/models.py +++ b/ESSArch_Core/WorkflowEngine/models.py @@ -113,7 +113,7 @@ class Meta: celery_id = models.UUIDField(default=uuid.uuid4, unique=True) name = models.CharField(max_length=255, db_index=True) label = models.CharField(max_length=255, blank=True) - queue = models.CharField(max_length=255, blank=True, null=True, default=None) + queue = models.CharField(max_length=255, blank=True, null=True, default='celery') hidden = models.BooleanField(null=True, default=None, db_index=True) eager = models.BooleanField(default=True) time_created = models.DateTimeField(auto_now_add=True) diff --git a/ESSArch_Core/WorkflowEngine/serializers.py b/ESSArch_Core/WorkflowEngine/serializers.py index 3ec3f5f19..7464a34b2 100644 --- a/ESSArch_Core/WorkflowEngine/serializers.py +++ b/ESSArch_Core/WorkflowEngine/serializers.py @@ -124,7 +124,7 @@ class Meta: 'processstep', 'processstep_pos', 'time_created', 'time_started', 'time_done', 'retried', 'responsible', 'hidden', 'args', 'params', 'information_package', - 'information_package_str', 'eager', + 'information_package_str', 'eager', 'queue', ) read_only_fields = ( 'time_created', 'time_started', 'time_done', 'retried', @@ -234,7 +234,7 @@ class Meta: 'url', 'id', 'name', 'label', 'result', 'part_root', 'user', 'parallel', 'run_state', 'status', 'progress', 'time_created', 'parent', 'parent_pos', 'information_package', 'information_package_str', - 'flow_type', 'step_position', 'responsible', + 'flow_type', 'step_position', 'responsible', 'queue', ) read_only_fields = ( 'status', 'progress', 'time_created', 'time_done', diff --git a/ESSArch_Core/WorkflowEngine/util.py b/ESSArch_Core/WorkflowEngine/util.py index 02dd2c715..7ee6b8dff 100644 --- a/ESSArch_Core/WorkflowEngine/util.py +++ b/ESSArch_Core/WorkflowEngine/util.py @@ -60,6 +60,7 @@ def _create_on_error_tasks(parent, errors, ip=None, responsible=None, eager=Fals processstep_pos=on_error_idx, status=status, progress=progress, + queue=on_error.get('queue') or parent.queue ) @@ -89,6 +90,7 @@ def _create_step(parent, flow, ip, responsible, context=None): information_package=ip, context=context, responsible=responsible, + queue=flow_entry.get('queue') or parent.queue, ) on_error_tasks = list(_create_on_error_tasks( @@ -111,7 +113,7 @@ def _create_step(parent, flow, ip, responsible, context=None): result_params = flow_entry.get('result_params', {}) task = ProcessTask.objects.create( name=name, - queue=flow_entry.get('queue', None), + queue=flow_entry.get('queue') or parent.queue, reference=flow_entry.get('reference', None), label=flow_entry.get('label'), args=args, @@ -157,6 +159,14 @@ def create_workflow(workflow_spec=None, ip=None, workflow_steps=None, name='', l logger = logging.getLogger('essarch.workflow') + queue = None + + if workflow_spec: + queue = workflow_spec[0].get('queue') + + if queue is None: + queue = context.get('WORKFLOW_QUEUE', 'celery') + with cache.lock('create_workflow_lock', timeout=300): try: for attempt in Retrying(stop=stop_after_delay(30), @@ -174,7 +184,7 @@ def create_workflow(workflow_spec=None, ip=None, workflow_steps=None, name='', l root_step = ProcessStep( name=name, eager=eager, information_package=ip, context=context, responsible=responsible, label=label, part_root=part_root, - run_state=run_state) + run_state=run_state, queue=queue) root_step.parent = top_root_step root_step.parent_pos = top_root_step.child_steps.count() + 1 root_step.save() @@ -182,7 +192,7 @@ def create_workflow(workflow_spec=None, ip=None, workflow_steps=None, name='', l root_step = ProcessStep.objects.create( name=name, eager=eager, information_package=ip, context=context, responsible=responsible, label=label, part_root=part_root, - run_state=run_state) + run_state=run_state, queue=queue) # Create on_error tasks on_error_tasks = list(_create_on_error_tasks( diff --git a/ESSArch_Core/config/settings.py b/ESSArch_Core/config/settings.py index e95b559de..4d12d8763 100644 --- a/ESSArch_Core/config/settings.py +++ b/ESSArch_Core/config/settings.py @@ -596,8 +596,15 @@ CELERY_TASK_TRACK_STARTED = True CELERY_WORKER_PREFETCH_MULTIPLIER = 1 CELERY_WORKER_CANCEL_LONG_RUNNING_TASKS_ON_CONNECTION_LOSS = True -ESSARCH_MAX_RUNNING_STEPS = 10 ESSARCH_RUN_SINGLE_ROOT_STEP_IF_RUNNING_PART_ROOT_STEP = True +ESSARCH_STEP_QUEUES = { + 'celery': { + 'max_running_steps': 10, + }, + 'poller': { + 'max_running_steps': 10, + } +} CELERY_BEAT_SCHEDULE = { 'RunWorkflowPollers-every-60-seconds': { diff --git a/ESSArch_Core/frontend/static/frontend/views/modals/step_info_modal.html b/ESSArch_Core/frontend/static/frontend/views/modals/step_info_modal.html index ed7b913aa..df13eb984 100644 --- a/ESSArch_Core/frontend/static/frontend/views/modals/step_info_modal.html +++ b/ESSArch_Core/frontend/static/frontend/views/modals/step_info_modal.html @@ -24,7 +24,7 @@
-

{{ currentStepTask.name }}

+

{{ currentStepTask.name }} ({{currentStepTask.queue}})

diff --git a/ESSArch_Core/frontend/static/frontend/views/modals/task_info_modal.html b/ESSArch_Core/frontend/static/frontend/views/modals/task_info_modal.html index acf493253..0946cdfea 100644 --- a/ESSArch_Core/frontend/static/frontend/views/modals/task_info_modal.html +++ b/ESSArch_Core/frontend/static/frontend/views/modals/task_info_modal.html @@ -24,7 +24,7 @@
-

{{ currentStepTask.name }}

+

{{ currentStepTask.name }} ({{currentStepTask.queue}})

diff --git a/ESSArch_Core/storage/serializers.py b/ESSArch_Core/storage/serializers.py index 4350e1d67..bf13519a3 100644 --- a/ESSArch_Core/storage/serializers.py +++ b/ESSArch_Core/storage/serializers.py @@ -494,7 +494,7 @@ def create(self, validated_data): media_migrate_workflow_step = ProcessStep.objects.create( name='Migrate Storage Medium', eager=False, information_package=None, context={}, responsible=user, label='Migrate Storage Medium {}'.format(medium_id), - part_root=None, run_state='') + part_root=None, run_state='', queue='celery') t = ProcessTask.objects.create( name='ESSArch_Core.storage.tasks.CreateMediumMigrationWorkflow', @@ -533,7 +533,7 @@ def create(self, validated_data): media_migrate_workflow_step = ProcessStep.objects.create( name='Migrate Storage Medium', eager=False, information_package=None, context={}, responsible=user, label='Migrate Storage Medium {}'.format(StorageMedium_obj.medium_id), - part_root=None, run_state='') + part_root=None, run_state='', queue='celery') t = ProcessTask.objects.create( name='ESSArch_Core.storage.tasks.CreateMediumMigrationWorkflow', diff --git a/ESSArch_Core/storage/tests/test_views.py b/ESSArch_Core/storage/tests/test_views.py index 45a12871e..f0cbaff44 100644 --- a/ESSArch_Core/storage/tests/test_views.py +++ b/ESSArch_Core/storage/tests/test_views.py @@ -1051,6 +1051,7 @@ def test_migration_task_order_for_information_packages(self, mock_task_run, mock label=mock.ANY, part_root=mock.ANY, run_state=mock.ANY, + queue=mock.ANY, )) for ip in [ips[0], ips[5], ips[3], ips[1], ips[2], ips[4]]: # calls.append(mock.call(name='Migrate Information Package', eager=False, information_package=ip, @@ -1058,19 +1059,19 @@ def test_migration_task_order_for_information_packages(self, mock_task_run, mock # run_state=mock.ANY)) calls.append(mock.call(name='Write to storage methods', parallel=mock.ANY, parent=mock.ANY, parent_pos=mock.ANY, eager=False, information_package=ip, context=mock.ANY, - responsible=mock.ANY)) + responsible=mock.ANY, queue=mock.ANY)) calls.append(mock.call(name='Write non-containers', parallel=mock.ANY, parent=mock.ANY, parent_pos=mock.ANY, eager=False, information_package=ip, context=mock.ANY, - responsible=mock.ANY)) + responsible=mock.ANY, queue=mock.ANY)) calls.append(mock.call(name='Write non-containers to storage methods', parallel=mock.ANY, parent=mock.ANY, parent_pos=mock.ANY, eager=False, information_package=ip, context=mock.ANY, - responsible=mock.ANY)) + responsible=mock.ANY, queue=mock.ANY)) calls.append(mock.call(name='Write containers to storage methods', parallel=mock.ANY, parent=mock.ANY, parent_pos=mock.ANY, eager=False, information_package=ip, context=mock.ANY, - responsible=mock.ANY)) + responsible=mock.ANY, queue=mock.ANY)) calls.append(mock.call(name='Delete temporary files', parallel=mock.ANY, parent=mock.ANY, parent_pos=mock.ANY, eager=False, information_package=ip, context=mock.ANY, - responsible=mock.ANY)) + responsible=mock.ANY, queue=mock.ANY)) mock_task.assert_has_calls(calls) @@ -1120,6 +1121,7 @@ def test_migration_task_order_for_storage_mediums(self, mock_task_run, mock_task label=mock.ANY, part_root=mock.ANY, run_state=mock.ANY, + queue=mock.ANY, )) for ip in [ips[0], ips[5], ips[3], ips[1], ips[2], ips[4]]: # calls.append(mock.call(name='Migrate Information Package', eager=False, information_package=ip, @@ -1127,19 +1129,19 @@ def test_migration_task_order_for_storage_mediums(self, mock_task_run, mock_task # run_state=mock.ANY)) calls.append(mock.call(name='Write to storage methods', parallel=mock.ANY, parent=mock.ANY, parent_pos=mock.ANY, eager=False, information_package=ip, context=mock.ANY, - responsible=mock.ANY)) + responsible=mock.ANY, queue=mock.ANY)) calls.append(mock.call(name='Write non-containers', parallel=mock.ANY, parent=mock.ANY, parent_pos=mock.ANY, eager=False, information_package=ip, context=mock.ANY, - responsible=mock.ANY)) + responsible=mock.ANY, queue=mock.ANY)) calls.append(mock.call(name='Write non-containers to storage methods', parallel=mock.ANY, parent=mock.ANY, parent_pos=mock.ANY, eager=False, information_package=ip, context=mock.ANY, - responsible=mock.ANY)) + responsible=mock.ANY, queue=mock.ANY)) calls.append(mock.call(name='Write containers to storage methods', parallel=mock.ANY, parent=mock.ANY, parent_pos=mock.ANY, eager=False, information_package=ip, context=mock.ANY, - responsible=mock.ANY)) + responsible=mock.ANY, queue=mock.ANY)) calls.append(mock.call(name='Delete temporary files', parallel=mock.ANY, parent=mock.ANY, parent_pos=mock.ANY, eager=False, information_package=ip, context=mock.ANY, - responsible=mock.ANY)) + responsible=mock.ANY, queue=mock.ANY)) mock_task.assert_has_calls(calls) diff --git a/ESSArch_Core/tasks.py b/ESSArch_Core/tasks.py index 13c7ff929..ac19d4e9a 100644 --- a/ESSArch_Core/tasks.py +++ b/ESSArch_Core/tasks.py @@ -823,9 +823,11 @@ def get_workflows(): backend = get_backend(name) poll_path = poller['path'] poll_sa = poller.get('sa') + poll_queue = poller.get('queue', 'poller') context = { 'WORKFLOW_POLLER': name, - 'WORKFLOW_POLL_PATH': poll_path + 'WORKFLOW_POLL_PATH': poll_path, + 'WORKFLOW_QUEUE': poll_queue } for ip in backend.poll(poll_path, poll_sa): profile = ip.submission_agreement.profile_workflow diff --git a/ESSArch_Core/workflow/tasks.py b/ESSArch_Core/workflow/tasks.py index 575d53fb6..9bada57a6 100644 --- a/ESSArch_Core/workflow/tasks.py +++ b/ESSArch_Core/workflow/tasks.py @@ -453,91 +453,311 @@ def UnmountIdleDrives(self): @app.task(bind=True, track=False) def PollProcessStepQueue(self): + """ + Poll ProcessStep queue. + + Each workflow lane (celery, poller, etc.) is handled independently. + + Example settings: + + ESSARCH_STEP_QUEUES = { + 'poller': {'max_running_steps': 10}, + 'celery': {'max_running_steps': 10}, + } + """ logger = logging.getLogger('essarch.workflow.tasks.PollProcessStepQueue') lock_key = 'poll_process_step_queue' - if cache.keys(lock_key): + + if cache.get(lock_key): logger.debug('Polling process step queue already running') return with cache.lock(lock_key, timeout=3600): + logger.debug('Polling process step queue') - cache_running_key = 'running_process_steps' - max_running_steps = getattr(settings, 'ESSARCH_MAX_RUNNING_STEPS', 10) - for root_step in ProcessStep.objects.filter(parent=None, run_state='STARTED'): - if (root_step.get_children().filter(part_root=True).exists() and - root_step.status not in [celery_states.FAILURE, celery_states.REVOKED]): - logger.info('root_step {} (STARTED) has part_root steps'.format(root_step)) - # Set run_state to SUCCESS if step.status is SUCCESS - for part_root_step in root_step.get_children().filter(part_root=True, run_state='STARTED'): - if part_root_step.status == celery_states.SUCCESS: - logger.info('root_step {} with part_root step {} is SUCCESS'.format(root_step, part_root_step)) - part_root_step.run_state = 'SUCCESS' - part_root_step.save(update_fields=['run_state']) - - # Run pending part root steps if max_running_steps is not reached - cache.set(cache_running_key, ProcessStep.objects.filter(run_state='STARTED').count()) - for part_root_step in root_step.get_children().filter(part_root=True, run_state='PENDING'): - if (part_root_step.status == celery_states.PENDING and - cache.get(cache_running_key) < max_running_steps): - logger.info('root_step {} with part_root step {} is PENDING starting'.format( - root_step, part_root_step)) - part_root_step.run() - part_root_step.run_state = 'STARTED' - part_root_step.save(update_fields=['run_state']) - cache.incr(cache_running_key) - else: - logger.info('root_step {} with part_root do not start more PENDING steps, already \ -running {}'.format(root_step, cache.get(cache_running_key))) - break - - # Set run_state to SUCCESS if no children are running and status is SUCCESS - if (not root_step.get_children().filter(part_root=True).exclude(run_state='SUCCESS').exists() and - root_step.status == celery_states.SUCCESS): - logger.info('root_step {} is SUCCESS'.format(root_step)) - root_step.run_state = 'SUCCESS' - root_step.save(update_fields=['run_state']) - cache.set(cache_running_key, ProcessStep.objects.filter(run_state='STARTED').count()) - if (ProcessStep.objects.filter(parent=None, run_state='PENDING').exists() and - cache.get(cache_running_key) < max_running_steps): - for root_step in ProcessStep.objects.filter(parent=None, run_state='PENDING'): - if root_step.get_children().filter(part_root=True).exists(): - if (getattr(settings, 'ESSARCH_RUN_SINGLE_ROOT_STEP_IF_RUNNING_PART_ROOT_STEP', True) and - ProcessStep.objects.filter(parent=None, run_state='STARTED', - child_steps__run_state__in=['STARTED', 'PENDING']).exists()): - logger.info('root_step with running part_root steps (STARTED or PENDING) already exists, do \ -not try to start new') - continue - logger.info('root_step {} (PENDING) has part_root steps'.format(root_step)) - # Run pending part root steps if max_running_steps is not reached - for part_root_step in root_step.get_children().filter(part_root=True, run_state='PENDING'): - if (part_root_step.status == celery_states.PENDING and - cache.get(cache_running_key) < max_running_steps): - logger.info('root_step {} with part_root step {} is PENDING starting'.format( - root_step, part_root_step)) - part_root_step.run() - part_root_step.run_state = 'STARTED' - part_root_step.save(update_fields=['run_state']) - cache.incr(cache_running_key) - else: - logger.info('root_step {} with part_root do not start more PENDING steps, already \ -running {}'.format(root_step, cache.get(cache_running_key))) - break - if (root_step.get_children().filter(part_root=True, run_state='STARTED').exists() and - root_step.run_state == 'PENDING'): - logger.info('root_step {} with STARTED part_root_step, flag root_step to STARTED'.format( - root_step)) - root_step.run_state = 'STARTED' - root_step.save(update_fields=['run_state']) - elif cache.get(cache_running_key) < max_running_steps: - # Run pending root steps if max_running_steps is not reached - logger.info('root_step {} without part_root_step starting and flag root_step to STARTED'.format( - root_step)) - root_step.run() - root_step.run_state = 'STARTED' - root_step.save(update_fields=['run_state']) + queue_config = getattr( + settings, + 'ESSARCH_STEP_QUEUES', + { + 'celery': { + 'max_running_steps': getattr( + settings, + 'ESSARCH_MAX_RUNNING_STEPS', + 10, + ) + }, + 'poller': { + 'max_running_steps': getattr( + settings, + 'ESSARCH_MAX_RUNNING_STEPS', + 10, + ) + } + } + ) + + for queue_name, cfg in queue_config.items(): + + max_running_steps = cfg.get( + 'max_running_steps', + getattr(settings, 'ESSARCH_MAX_RUNNING_STEPS', 10) + ) + + logger.debug( + 'Processing workflow queue "%s" (max_running_steps=%s)', + queue_name, + max_running_steps, + ) + + _poll_workflow_queue( + queue_name, + max_running_steps, + logger, + ) + + +def _poll_workflow_queue(queue_name, max_running_steps, logger): + + cache_running_key = f'running_process_steps_{queue_name}' + + # + # Handle workflows already running + # + + for root_step in ProcessStep.objects.filter( + parent=None, + queue=queue_name, + run_state='STARTED', + ): + + # + # Root workflow with part_root children + # + + if ( + root_step.get_children().filter(part_root=True).exists() and + root_step.status not in [ + celery_states.FAILURE, + celery_states.REVOKED, + ] + ): + + logger.info( + 'root_step %s (STARTED) has part_root steps [queue=%s]', + root_step, + queue_name, + ) + + # + # Mark completed part_root steps + # + + for part_root_step in root_step.get_children().filter( + part_root=True, + run_state='STARTED', + ): + + if part_root_step.status == celery_states.SUCCESS: + + logger.info( + 'part_root_step %s SUCCESS [queue=%s]', + part_root_step, + queue_name, + ) + + part_root_step.run_state = 'SUCCESS' + part_root_step.save(update_fields=['run_state']) + + # + # Start new part_root steps if capacity exists + # + + running_count = ProcessStep.objects.filter( + queue=queue_name, + run_state='STARTED', + ).count() + + cache.set(cache_running_key, running_count) + + for part_root_step in root_step.get_children().filter( + part_root=True, + run_state='PENDING', + ): + + if ( + part_root_step.status == celery_states.PENDING and + cache.get(cache_running_key) < max_running_steps + ): + + logger.info( + 'Starting part_root_step %s [queue=%s]', + part_root_step, + queue_name, + ) + + part_root_step.run() + + part_root_step.run_state = 'STARTED' + part_root_step.save(update_fields=['run_state']) + cache.incr(cache_running_key) + + else: + + logger.info( + 'Queue %s full (%s/%s)', + queue_name, + cache.get(cache_running_key), + max_running_steps, + ) + + break + + # + # Mark root workflow complete + # + + if ( + not root_step.get_children() + .filter(part_root=True) + .exclude(run_state='SUCCESS') + .exists() and + root_step.status == celery_states.SUCCESS + ): + + logger.info( + 'root_step %s SUCCESS [queue=%s]', + root_step, + queue_name, + ) + + root_step.run_state = 'SUCCESS' + root_step.save(update_fields=['run_state']) + + # + # Recalculate running count + # + + running_count = ProcessStep.objects.filter( + queue=queue_name, + run_state='STARTED', + ).count() + + cache.set(cache_running_key, running_count) + + # + # Start pending root workflows + # + + pending_root_steps = ProcessStep.objects.filter( + parent=None, + queue=queue_name, + run_state='PENDING', + ).order_by('time_created') + + for root_step in pending_root_steps: + + if cache.get(cache_running_key) >= max_running_steps: + + logger.info( + 'Queue %s full (%s/%s)', + queue_name, + cache.get(cache_running_key), + max_running_steps, + ) + + break + + # + # Root workflow contains part_root lanes + # + + if root_step.get_children().filter(part_root=True).exists(): + + # + # Preserve existing behaviour: + # only one root workflow with active part_root execution + # + + if ( + getattr( + settings, + 'ESSARCH_RUN_SINGLE_ROOT_STEP_IF_RUNNING_PART_ROOT_STEP', + True, + ) and + ProcessStep.objects.filter( + parent=None, + queue=queue_name, + run_state='STARTED', + child_steps__run_state__in=['STARTED', 'PENDING'], + ).exists() + ): + + logger.info( + 'Queue %s already has active root workflow with part_root steps', + queue_name, + ) + + continue + + logger.info( + 'root_step %s has part_root steps [queue=%s]', + root_step, + queue_name, + ) + + for part_root_step in root_step.get_children().filter( + part_root=True, + run_state='PENDING', + ): + + if ( + part_root_step.status == celery_states.PENDING and + cache.get(cache_running_key) < max_running_steps + ): + + logger.info( + 'Starting part_root_step %s [queue=%s]', + part_root_step, + queue_name, + ) + + part_root_step.run() + + part_root_step.run_state = 'STARTED' + part_root_step.save(update_fields=['run_state']) + + cache.incr(cache_running_key) + else: - logger.info('Do not start more PENDING steps, already running {}'.format( - cache.get(cache_running_key))) break + + if ( + root_step.get_children() + .filter(part_root=True, run_state='STARTED') + .exists() and + root_step.run_state == 'PENDING' + ): + + root_step.run_state = 'STARTED' + root_step.save(update_fields=['run_state']) + + # + # Standard workflow + # + + else: + + logger.info( + 'Starting root_step %s [queue=%s]', + root_step, + queue_name, + ) + + root_step.run() + + root_step.run_state = 'STARTED' + root_step.save(update_fields=['run_state']) + + cache.incr(cache_running_key)