Merge branch 'api-manage-tasks' of 'https://github.com/jjmerchante/grimoirelab-core'

Merges #35
Closes #35

Author: sduenas

src/grimoirelab/core/scheduler/scheduler.py

@@ -29,6 +29,7 @@
 
 from django.conf import settings
 from rq.registry import StartedJobRegistry
+from rq.command import send_stop_job_command
 
 from grimoirelab_toolkit.datetime import datetime_utcnow
 
@@ -104,16 +105,53 @@ def cancel_task(task_uuid: str) -> None:
     """
     task = find_task(task_uuid)
 
-    _, job_class = get_registered_task_model(task.TASK_TYPE)
-
-    jobs = job_class.objects.filter(task=task).all()
+    jobs = task.jobs.all()
     for job in jobs:
-        job_rq = rq.job.Job.fetch(job.uuid, connection=django_rq.get_connection(task.default_job_queue))
+        connection = django_rq.get_connection(task.default_job_queue)
+        try:
+            job_rq = rq.job.Job.fetch(job.uuid, connection=connection)
+        except rq.exceptions.NoSuchJobError:
+            continue
+        if job_rq.get_status() == rq.job.JobStatus.STARTED:
+            send_stop_job_command(connection, job_rq.id)
         job_rq.delete()
 
     task.delete()
 
 
+def reschedule_task(task_uuid: str) -> None:
+    """Reschedule a task
+
+    The task will be rescheduled to be executed as soon
+    as possible. If it is running, it will be cancelled and
+    rescheduled.
+
+    :param task_uuid: uuid of the task to be rescheduled.
+
+    :raises NotFoundError: when the task is not found.
+    """
+    task = find_task(task_uuid)
+
+    if task.status == SchedulerStatus.ENQUEUED:
+        # Cancel the enqueued job and force the execution
+        job = task.jobs.order_by('-scheduled_at').first()
+        try:
+            job_rq = rq.job.Job.fetch(job.uuid, connection=django_rq.get_connection(task.default_job_queue))
+            job_rq.delete()
+        except (rq.exceptions.NoSuchJobError, rq.exceptions.InvalidJobOperation):
+            pass
+        _schedule_job(task, job, datetime_utcnow(), job.job_args)
+
+    elif task.status == SchedulerStatus.RUNNING:
+        # Make sure it is running
+        job = task.jobs.order_by('-scheduled_at').first()
+        if _is_job_removed_or_stopped(job, task.default_job_queue):
+            _schedule_job(task, job, datetime_utcnow(), job.job_args)
+
+    else:
+        _enqueue_task(task)
+
+
 def maintain_tasks() -> None:
     """Maintain the tasks that are scheduled to be executed.
 

src/grimoirelab/core/scheduler/urls.py

@@ -24,6 +24,8 @@
 
 urlpatterns = [
     re_path(r'^add_task', views.add_task),
+    re_path(r'^reschedule_task', views.reschedule_task),
+    re_path(r'^remove_task', views.remove_task),
     path('tasks/', api.EventizerTaskList.as_view()),
     path('tasks/<str:uuid>/', api.EventizerTaskDetail.as_view()),
     path('tasks/<str:task_id>/jobs/', api.EventizerJobList.as_view()),

src/grimoirelab/core/scheduler/views.py

@@ -21,7 +21,9 @@
 from django.conf import settings
 
 from .scheduler import (
-    schedule_task
+    cancel_task,
+    schedule_task,
+    reschedule_task as scheduler_reschedule_task
 )
 
 
@@ -71,3 +73,45 @@ def add_task(request):
         'message': f"Task {task.id} added correctly"
     }
     return Response(response, status=200)
+
+
+@api_view(['POST'])
+def reschedule_task(request):
+    """Reschedule a Task
+
+    The body should contain the task id to reschedule:
+    {
+        'taskId': 'task_id'
+    }
+    """
+    data = request.data
+    task_id = data['taskId']
+
+    scheduler_reschedule_task(task_id)
+
+    response = {
+        'status': 'ok',
+        'message': f"Task {task_id} rescheduled correctly"
+    }
+    return Response(response, status=200)
+
+
+@api_view(['POST'])
+def remove_task(request):
+    """Remove a Task
+
+    The body should contain the task id to remove:
+    {
+        'taskId': 'task_id'
+    }
+    """
+    data = request.data
+    task_id = data['taskId']
+
+    cancel_task(task_id)
+
+    response = {
+        'status': 'ok',
+        'message': f"Task {task_id} removed correctly"
+    }
+    return Response(response, status=200)

tests/scheduler/test_scheduler.py

@@ -36,6 +36,7 @@
     schedule_task,
     cancel_task,
     maintain_tasks,
+    reschedule_task,
     _enqueue_task,
     _on_success_callback,
     _on_failure_callback
@@ -511,6 +512,156 @@ def test_no_task_found(self):
             cancel_task('non-existent-task-uuid')
 
 
+class TestRescheduleTask(GrimoireLabTestCase):
+    def setUp(self):
+        GRIMOIRELAB_TASK_MODELS.clear()
+        task_class, job_class = register_task_model('test_task', SchedulerTestTask)
+
+        def cleanup_test_model():
+            with django.db.connection.schema_editor() as schema_editor:
+                schema_editor.delete_model(job_class)
+                schema_editor.delete_model(task_class)
+
+        with django.db.connection.schema_editor() as schema_editor:
+            schema_editor.create_model(task_class)
+            schema_editor.create_model(job_class)
+
+        self.addCleanup(cleanup_test_model)
+        super().setUp()
+
+    def test_reschedule_task_completed(self):
+        """Test a task is rescheduled correctly"""
+
+        task_args = {
+            'a': 1,
+            'b': 2,
+        }
+
+        # Enqueue the task
+        enqueued_at = grimoirelab_toolkit.datetime.datetime_utcnow()
+        task = schedule_task('test_task', task_args)
+
+        # Check if a new job is created and the task is enqueued
+        self.assertEqual(task.status, SchedulerStatus.ENQUEUED)
+        self.assertGreaterEqual(task.scheduled_at, enqueued_at)
+
+        # Run the job
+        before_run_call_dt = grimoirelab_toolkit.datetime.datetime_utcnow()
+        worker = django_rq.workers.get_worker('testing')
+        processed = worker.work(burst=True, with_scheduler=True)
+        after_run_call_dt = grimoirelab_toolkit.datetime.datetime_utcnow()
+
+        self.assertEqual(processed, True)
+
+        # Check task and task state after execution
+        task.refresh_from_db()
+        self.assertEqual(task.status, SchedulerStatus.COMPLETED)
+        self.assertEqual(task.runs, 1)
+        self.assertEqual(task.failures, 0)
+        self.assertGreater(task.last_run, before_run_call_dt)
+        self.assertLess(task.last_run, after_run_call_dt)
+
+        # Reschedule the completed task
+        enqueued_at = grimoirelab_toolkit.datetime.datetime_utcnow()
+        reschedule_task(task.uuid)
+
+        # Check if the task is rescheduled
+        task.refresh_from_db()
+        self.assertEqual(task.status, SchedulerStatus.ENQUEUED)
+        self.assertGreater(task.scheduled_at, enqueued_at)
+
+        # Run the job
+        before_run_call_dt = grimoirelab_toolkit.datetime.datetime_utcnow()
+        worker = django_rq.workers.get_worker('testing')
+        processed = worker.work(burst=True, with_scheduler=True)
+        after_run_call_dt = grimoirelab_toolkit.datetime.datetime_utcnow()
+
+        self.assertEqual(processed, True)
+
+        # Check task and task state after execution
+        task.refresh_from_db()
+        self.assertEqual(task.status, SchedulerStatus.COMPLETED)
+        self.assertEqual(task.runs, 2)
+        self.assertEqual(task.failures, 0)
+        self.assertGreater(task.last_run, before_run_call_dt)
+        self.assertLess(task.last_run, after_run_call_dt)
+
+    def test_reschedule_task_enqueued(self):
+        """Test if rescheduling a task enqueued change the scheduled time"""
+
+        # Enqueue the task
+        task_args = {
+            'a': 1,
+            'b': 2,
+        }
+        first_schedule_time = datetime.datetime(2100, 1, 1, tzinfo=datetime.timezone.utc)
+
+        task = SchedulerTestTask.create_task(task_args, 360, 10)
+        job_db = _enqueue_task(task, scheduled_at=first_schedule_time)
+
+        # Check initial state of the task
+        self.assertEqual(task.task_id, f'grimoire:task:{task.uuid}')
+        self.assertEqual(task.status, SchedulerStatus.ENQUEUED)
+        self.assertEqual(task.scheduled_at, first_schedule_time)
+
+        # Reschedule the task
+        reschedule_task(task.uuid)
+        second_schedule_time = grimoirelab_toolkit.datetime.datetime_utcnow()
+
+        # Check if the task is rescheduled
+        task.refresh_from_db()
+        self.assertEqual(task.status, SchedulerStatus.ENQUEUED)
+        self.assertLess(task.scheduled_at, second_schedule_time)
+
+    def test_reschedule_inconsistent_task(self):
+        """Test if reschedules a task with an inconsistent state"""
+
+        # Enqueue the task
+        task_args = {
+            'a': 1,
+            'b': 2,
+        }
+
+        task = SchedulerTestTask.create_task(task_args, 360, 10)
+        job_db = _enqueue_task(task)
+
+        # Check initial state of the task
+        self.assertEqual(task.status, SchedulerStatus.ENQUEUED)
+
+        # Delete job manually to create the inconsistent state
+        job_rq = rq.job.Job.fetch(job_db.uuid, connection=django_rq.get_connection())
+        job_rq.delete()
+
+        # Make sure the job was deleted
+        with self.assertRaises(rq.exceptions.NoSuchJobError):
+            rq.job.Job.fetch(job_db.uuid, connection=django_rq.get_connection())
+
+        # Reschedule the task
+        scheduled_date = grimoirelab_toolkit.datetime.datetime_utcnow()
+        reschedule_task(task.uuid)
+
+        # Check if the task is rescheduled
+        task.refresh_from_db()
+        job_db = task.jobs.order_by('-scheduled_at').first()
+
+        self.assertEqual(task.status, SchedulerStatus.ENQUEUED)
+        self.assertEqual(job_db.status, SchedulerStatus.ENQUEUED)
+        self.assertGreater(task.scheduled_at, scheduled_date)
+
+        # Run the job
+        worker = django_rq.workers.get_worker('testing')
+        processed = worker.work(burst=True, with_scheduler=True)
+
+        self.assertEqual(processed, True)
+
+        # Check task and task state after execution
+        task.refresh_from_db()
+        self.assertEqual(task.status, SchedulerStatus.COMPLETED)
+        self.assertEqual(task.runs, 1)
+        self.assertEqual(task.failures, 0)
+        self.assertGreater(task.last_run, scheduled_date)
+
+
 class TestOnSuccessCallback(GrimoireLabTestCase):
     """Unit tests for the default on_success_callback function"""