Merge pull request #4961 from jmchilton/job_delete_api

Add DELETE /api/jobs/<job_id> as a job cancellation API endpoint.

lib/galaxy/webapps/galaxy/api/jobs.py

@@ -171,7 +171,7 @@ class JobController(BaseAPIController, UsesLibraryMixinItems):
     @expose_api
     def outputs(self, trans, id, **kwd):
         """
-        show( trans, id )
+        outputs( trans, id )
         * GET /api/jobs/{id}/outputs
             returns output datasets created by job
 
@@ -184,6 +184,25 @@ class JobController(BaseAPIController, UsesLibraryMixinItems):
         job = self.__get_job(trans, id)
         return self.__dictify_associations(trans, job.output_datasets, job.output_library_datasets)
 
+    @expose_api
+    def delete(self, trans, id, **kwd):
+        """
+        delete( trans, id )
+        * Delete /api/jobs/{id}
+            cancels specified job
+
+        :type   id: string
+        :param  id: Encoded job id
+        """
+        job = self.__get_job(trans, id)
+        if not job.finished:
+            job.mark_deleted(self.app.config.track_jobs_in_database)
+            trans.sa_session.flush()
+            self.app.job_manager.job_stop_queue.put(job.id)
+            return True
+        else:
+            return False
+
     @expose_api_anonymous
     def build_for_rerun(self, trans, id, **kwd):
         """

test/base/populators.py

@@ -157,6 +157,9 @@ class BaseDatasetPopulator(object):
     def get_job_details(self, job_id, full=False):
         return self._get("jobs/%s?full=%s" % (job_id, full))
 
+    def cancel_job(self, job_id):
+        return self._delete("jobs/%s" % job_id)
+
     def _summarize_history(self, history_id):
         pass
 
@@ -306,6 +309,9 @@ class DatasetPopulator(BaseDatasetPopulator):
     def _get(self, route, data={}):
         return self.galaxy_interactor.get(route, data=data)
 
+    def _delete(self, route, data={}):
+        return self.galaxy_interactor.delete(route, data=data)
+
     def _summarize_history(self, history_id):
         self.galaxy_interactor._summarize_history(history_id)
 
@@ -636,6 +642,11 @@ class GiPostGetMixin:
         data['key'] = self._gi.key
         return requests.post(self.__url(route), data=data)
 
+    def _delete(self, route, data={}):
+        data = data.copy()
+        data['key'] = self._gi.key
+        return requests.delete(self.__url(route), data=data)
+
     def __url(self, route):
         return self._gi.url + "/" + route
 

test/functional/tools/cat_data_and_sleep.xml

@@ -0,0 +1,21 @@
+<tool id="cat_data_and_sleep" name="Concatenate datasets (with sleep)" version="0.1.0">
+    <description>tail-to-head</description>
+    <command><![CDATA[
+        cat $input1 #for $q in $queries# ${q.input2} #end for# > $out_file1;
+        sleep '$sleep_time';
+    ]]></command>
+    <inputs>
+        <param name="sleep_time" type="integer" label="Sleep" help="Optionally simulates computation before concatenating files" value="0" />
+        <param name="input1" type="data" label="Concatenate Dataset"/>
+        <repeat name="queries" title="Dataset">
+            <param name="input2" type="data" label="Select" />
+        </repeat>
+    </inputs>
+    <outputs>
+        <data name="out_file1" format_source="input1" metadata_source="input1"/>
+    </outputs>
+    <tests>
+    </tests>
+    <help>
+    </help>
+</tool>

test/functional/tools/samples_tool_conf.xml

@@ -153,6 +153,9 @@
 
   <tool file="simple_constructs.yml" />
 
+  <!-- Tools without tool test but useful for hand-crafted, artisanal test cases. -->
+  <tool file="cat_data_and_sleep.xml" />
+
   <!-- Load collection operation tools - I consider these part of the
        "framework" and they are used to in API tests to test the underlying
        ToolActions. -->

test/galaxy_selenium/navigates_galaxy.py

@@ -172,7 +172,7 @@ class NavigatesGalaxy(HasDriver):
 
     def api_delete(self, endpoint, raw=False):
         full_url = self.build_url("api/" + endpoint, for_selenium=False)
-        response = requests.get(full_url, cookies=self.selenium_to_requests_cookies())
+        response = requests.delete(full_url, cookies=self.selenium_to_requests_cookies())
         if raw:
             return response
         else:

test/integration/test_local_job_cancellation.py

@@ -0,0 +1,80 @@
+"""Integration test for the local job runner and cancelling jobs via API."""
+
+import time
+
+import psutil
+
+from base import integration_util
+from base.populators import (
+    DatasetPopulator,
+)
+
+
+class LocalJobCancellationTestCase(integration_util.IntegrationTestCase):
+
+    framework_tool_and_types = True
+
+    def setUp(self):
+        super(LocalJobCancellationTestCase, self).setUp()
+        self.dataset_populator = DatasetPopulator(self.galaxy_interactor)
+
+    def test_kill_process(self):
+        """
+        """
+        with self.dataset_populator.test_history() as history_id:
+            hda1 = self.dataset_populator.new_dataset(history_id, content="1 2 3")
+            running_inputs = {
+                "input1": {"src": "hda", "id": hda1["id"]},
+                "sleep_time": 240,
+            }
+            running_response = self.dataset_populator.run_tool(
+                "cat_data_and_sleep",
+                running_inputs,
+                history_id,
+                assert_ok=False,
+            ).json()
+            job_dict = running_response["jobs"][0]
+
+            app = self._app
+            sa_session = app.model.context.current
+            external_id = None
+            state = False
+
+            job = sa_session.query(app.model.Job).filter_by(tool_id="cat_data_and_sleep").one()
+            # Not checking the state here allows the change from queued to running to overwrite
+            # the change from queued to deleted_new in the API thread - this is a problem because
+            # the job will still run. See issue https://github.com/galaxyproject/galaxy/issues/4960.
+            while external_id is None or state != app.model.Job.states.RUNNING:
+                sa_session.refresh(job)
+                assert not job.finished
+                external_id = job.job_runner_external_id
+                state = job.state
+
+            assert external_id
+            external_id = int(external_id)
+
+            pid_exists = psutil.pid_exists(external_id)
+            assert pid_exists
+
+            delete_response = self.dataset_populator.cancel_job(job_dict["id"])
+            assert delete_response.json() is True
+
+            state = None
+            # Now make sure the job becomes complete.
+            for i in range(100):
+                sa_session.refresh(job)
+                state = job.state
+                if state == app.model.Job.states.DELETED:
+                    break
+                time.sleep(.1)
+
+            # Now make sure the pid is actually killed.
+            for i in range(100):
+                if not pid_exists:
+                    break
+                pid_exists = psutil.pid_exists(external_id)
+                time.sleep(.1)
+
+            final_state = "pid exists? %s, final db job state %s" % (pid_exists, state)
+            assert state == app.model.Job.states.DELETED, final_state
+            assert not pid_exists, final_state