From 140a0a49d3af5f5d6ffaa50cc0cac5b5f4cd1bcc Mon Sep 17 00:00:00 2001 From: John Chilton Date: Mon, 6 Nov 2017 15:16:48 -0500 Subject: [PATCH] Add DELETE /api/jobs/ as a job cancellation API endpoint. --- lib/galaxy/webapps/galaxy/api/jobs.py | 21 ++++- test/base/populators.py | 11 +++ test/functional/tools/cat_data_and_sleep.xml | 21 +++++ test/functional/tools/samples_tool_conf.xml | 3 + test/galaxy_selenium/navigates_galaxy.py | 2 +- .../test_local_job_cancellation.py | 80 +++++++++++++++++++ 6 files changed, 136 insertions(+), 2 deletions(-) create mode 100644 test/functional/tools/cat_data_and_sleep.xml create mode 100644 test/integration/test_local_job_cancellation.py diff --git a/lib/galaxy/webapps/galaxy/api/jobs.py b/lib/galaxy/webapps/galaxy/api/jobs.py index 86f1ef415cb..75af1d5d5fb 100644 --- a/lib/galaxy/webapps/galaxy/api/jobs.py +++ b/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): """ diff --git a/test/base/populators.py b/test/base/populators.py index 46a9557ccab..dc664a9fc19 100644 --- a/test/base/populators.py +++ b/test/base/populators.py @@ -138,6 +138,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 @@ -287,6 +290,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) @@ -617,6 +623,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 diff --git a/test/functional/tools/cat_data_and_sleep.xml b/test/functional/tools/cat_data_and_sleep.xml new file mode 100644 index 00000000000..0cc6efef2c1 --- /dev/null +++ b/test/functional/tools/cat_data_and_sleep.xml @@ -0,0 +1,21 @@ + + tail-to-head + $out_file1; + sleep '$sleep_time'; + ]]> + + + + + + + + + + + + + + + diff --git a/test/functional/tools/samples_tool_conf.xml b/test/functional/tools/samples_tool_conf.xml index c35229e948b..0e1fd85f940 100644 --- a/test/functional/tools/samples_tool_conf.xml +++ b/test/functional/tools/samples_tool_conf.xml @@ -153,6 +153,9 @@ + + + diff --git a/test/galaxy_selenium/navigates_galaxy.py b/test/galaxy_selenium/navigates_galaxy.py index 01c11f7e8dc..59b12c57bba 100644 --- a/test/galaxy_selenium/navigates_galaxy.py +++ b/test/galaxy_selenium/navigates_galaxy.py @@ -173,7 +173,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: diff --git a/test/integration/test_local_job_cancellation.py b/test/integration/test_local_job_cancellation.py new file mode 100644 index 00000000000..65dfb42de89 --- /dev/null +++ b/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