diff --git a/lib/galaxy/jobs/__init__.py b/lib/galaxy/jobs/__init__.py index 7d9651e883d..ad7575e1157 100644 --- a/lib/galaxy/jobs/__init__.py +++ b/lib/galaxy/jobs/__init__.py @@ -913,9 +913,9 @@ class JobWrapper(HasResourceParameters): job.id) def _create_working_directory(self, job): - self.app.object_store.create( + self.object_store.create( job, base_dir='job_work', dir_only=True, obj_dir=True) - working_directory = self.app.object_store.get_filename( + working_directory = self.object_store.get_filename( job, base_dir='job_work', dir_only=True, obj_dir=True) return working_directory @@ -928,10 +928,10 @@ class JobWrapper(HasResourceParameters): self.working_directory) return - self.app.object_store.create( + self.object_store.create( job, base_dir='job_work', dir_only=True, obj_dir=True, extra_dir='_cleared_contents', extra_dir_at_root=True) - base = self.app.object_store.get_filename( + base = self.object_store.get_filename( job, base_dir='job_work', dir_only=True, obj_dir=True, extra_dir='_cleared_contents', extra_dir_at_root=True) date_str = datetime.datetime.now().strftime('%Y%m%d-%H%M%S') @@ -1169,11 +1169,12 @@ class JobWrapper(HasResourceParameters): def enqueue(self): job = self.get_job() - self._set_object_store_ids(job) # Change to queued state before handing to worker thread so the runner won't pick it up again self.change_state(model.Job.states.QUEUED, flush=False, job=job) # Persist the destination so that the job will be included in counts if using concurrency limits self.set_job_destination(self.job_destination, None, flush=False, job=job) + # Set object store after job destination so can leverage parameters... + self._set_object_store_ids(job) self.sa_session.flush() def _set_object_store_ids(self, job): @@ -1186,6 +1187,9 @@ class JobWrapper(HasResourceParameters): return object_store_populator = ObjectStorePopulator(self.app) + object_store_id = self.get_destination_configuration("object_store_id", None) + if object_store_id: + object_store_populator.object_store_id = object_store_id # Ideally we would do this without loading the actual job association # objects but change_state isn't yet optimized to do that so we need to @@ -1316,7 +1320,7 @@ class JobWrapper(HasResourceParameters): dataset.dataset.uuid = context['uuid'] # Update (non-library) job output datasets through the object store if dataset not in job.output_library_datasets: - self.app.object_store.update_from_file(dataset.dataset, create=True) + self.object_store.update_from_file(dataset.dataset, create=True) self.__update_output(job, dataset) if not purged: self._collect_extra_files(dataset.dataset, self.working_directory) @@ -1326,7 +1330,7 @@ class JobWrapper(HasResourceParameters): with NamedTemporaryFile() as temp_fh: temp_fh.write(dataset.datatype.generate_primary_file(dataset)) temp_fh.flush() - self.app.object_store.update_from_file(dataset.dataset, file_name=temp_fh.name, create=True) + self.object_store.update_from_file(dataset.dataset, file_name=temp_fh.name, create=True) dataset.set_size() except Exception as e: log.warning('Unable to generate primary composite file automatically for %s: %s', dataset.dataset.id, e) @@ -1522,7 +1526,7 @@ class JobWrapper(HasResourceParameters): galaxy.tools.imp_exp.JobExportHistoryArchiveWrapper(self.job_id).cleanup_after_job(self.sa_session) galaxy.tools.imp_exp.JobImportHistoryArchiveWrapper(self.app, self.job_id).cleanup_after_job() if delete_files: - self.app.object_store.delete(self.get_job(), base_dir='job_work', entire_dir=True, dir_only=True, obj_dir=True) + self.object_store.delete(self.get_job(), base_dir='job_work', entire_dir=True, dir_only=True, obj_dir=True) except Exception: log.exception("Unable to cleanup job %d", self.job_id) @@ -1537,7 +1541,7 @@ class JobWrapper(HasResourceParameters): for root, dirs, files in os.walk(temp_file_path): extra_dir = root.replace(job_working_directory, '', 1).lstrip(os.path.sep) for f in files: - self.app.object_store.update_from_file( + self.object_store.update_from_file( dataset, extra_dir=extra_dir, alt_name=f, @@ -1682,6 +1686,10 @@ class JobWrapper(HasResourceParameters): return dp.dataset_id return None + @property + def object_store(self): + return self.app.object_store + @property def tmp_dir_creation_statement(self): tmp_dir = self.get_destination_configuration("tmp_dir", None) @@ -1815,7 +1823,7 @@ class JobWrapper(HasResourceParameters): if dataset not in job.output_library_datasets: purged = dataset.purged if not purged and not clean_only: - self.app.object_store.update_from_file(dataset, create=True) + self.object_store.update_from_file(dataset, create=True) else: # If the dataset is purged and Galaxy is configured to write directly # to the object store from jobs - be sure that file is cleaned up. This diff --git a/lib/galaxy/objectstore/__init__.py b/lib/galaxy/objectstore/__init__.py index 62eef775eb7..db5103650b7 100644 --- a/lib/galaxy/objectstore/__init__.py +++ b/lib/galaxy/objectstore/__init__.py @@ -632,7 +632,7 @@ class DistributedObjectStore(NestedObjectStore): def create(self, obj, **kwargs): """The only method in which obj.object_store_id may be None.""" if obj.object_store_id is None or not self.exists(obj, **kwargs): - if obj.object_store_id is None or obj.object_store_id not in self.weighted_backend_ids: + if obj.object_store_id is None or obj.object_store_id not in self.backends: try: obj.object_store_id = random.choice(self.weighted_backend_ids) except IndexError: diff --git a/test/integration/objectstore_selection_job_conf.xml b/test/integration/objectstore_selection_job_conf.xml new file mode 100644 index 00000000000..9973e095dea --- /dev/null +++ b/test/integration/objectstore_selection_job_conf.xml @@ -0,0 +1,47 @@ + + + + + + + integration.objectstore_selection_rules + + + + + + + + + + + + + + + static + + + + + python + the_destination + + + + + + + how_store + + + + + + + + + diff --git a/test/integration/objectstore_selection_job_resource_parameters_conf.xml b/test/integration/objectstore_selection_job_resource_parameters_conf.xml new file mode 100644 index 00000000000..bbdbf1351b3 --- /dev/null +++ b/test/integration/objectstore_selection_job_resource_parameters_conf.xml @@ -0,0 +1,6 @@ + + + + + + diff --git a/test/integration/objectstore_selection_rules/__init__.py b/test/integration/objectstore_selection_rules/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/test/integration/objectstore_selection_rules/rules.py b/test/integration/objectstore_selection_rules/rules.py new file mode 100644 index 00000000000..ded02fc33fe --- /dev/null +++ b/test/integration/objectstore_selection_rules/rules.py @@ -0,0 +1,14 @@ +from galaxy.jobs import JobDestination + + +def the_destination(resource_params): + job_destination = JobDestination() + job_destination.runner = "local" + how_store = resource_params.get("how_store", None) + # Retrieve answer from user about whether to store on fast or slow disk, + # translate to object store ID to give to job handling code. + object_store_id = "dynamic_ebs" + if how_store == "slow": + object_store_id = "dynamic_s3" + job_destination.params['object_store_id'] = object_store_id + return job_destination diff --git a/test/integration/test_objectstore_selection.py b/test/integration/test_objectstore_selection.py new file mode 100644 index 00000000000..a6f25eb8cf6 --- /dev/null +++ b/test/integration/test_objectstore_selection.py @@ -0,0 +1,128 @@ +"""Integration tests for object stores.""" + +import os +import string + +from base import integration_util # noqa: I202 +from base.populators import ( + DatasetPopulator, +) + +SCRIPT_DIRECTORY = os.path.abspath(os.path.dirname(__file__)) +JOB_CONFIG_FILE = os.path.join(SCRIPT_DIRECTORY, "objectstore_selection_job_conf.xml") +JOB_RESOURCE_PARAMETERS_CONFIG_FILE = os.path.join(SCRIPT_DIRECTORY, "objectstore_selection_job_resource_parameters_conf.xml") + +DISTRIBUTED_OBJECT_STORE_CONFIG_TEMPLATE = string.Template(""" + + + + + + + + + + + + + + + + + + + + + + + + +""") + + +class ObjectStoreJobsIntegrationTestCase(integration_util.IntegrationTestCase): + + framework_tool_and_types = True + + @classmethod + def handle_galaxy_config_kwds(cls, config): + temp_directory = cls._test_driver.mkdtemp() + cls.object_stores_parent = temp_directory + for disk_store_file_name in ["files_default", "files_static", "files_dynamic_ebs", "files_dynamic_s3"]: + disk_store_path = os.path.join(temp_directory, disk_store_file_name) + os.makedirs(disk_store_path) + setattr(cls, "%s_path" % disk_store_file_name, disk_store_path) + config_path = os.path.join(temp_directory, "object_store_conf.xml") + with open(config_path, "w") as f: + f.write(DISTRIBUTED_OBJECT_STORE_CONFIG_TEMPLATE.safe_substitute({"temp_directory": temp_directory})) + config["object_store_config_file"] = config_path + config["job_config_file"] = JOB_CONFIG_FILE + config["job_resource_params_file"] = JOB_RESOURCE_PARAMETERS_CONFIG_FILE + + def setUp(self): + super(ObjectStoreJobsIntegrationTestCase, self).setUp() + self.dataset_populator = DatasetPopulator(self.galaxy_interactor) + + def _object_store_counts(self): + files_default_count = _files_count(self.files_default_path) + files_static_count = _files_count(self.files_static_path) + files_dynamic_count = _files_count(self.files_dynamic_path) + return files_default_count, files_static_count, files_dynamic_count + + def _assert_file_counts(self, default, static, dynamic_ebs, dynamic_s3): + files_default_count = _files_count(self.files_default_path) + files_static_count = _files_count(self.files_static_path) + files_dynamic_ebs_count = _files_count(self.files_dynamic_ebs_path) + files_dynamic_s3_count = _files_count(self.files_dynamic_s3_path) + assert default == files_default_count + assert static == files_static_count + assert dynamic_ebs == files_dynamic_ebs_count + assert dynamic_s3 == files_dynamic_s3_count + + def test_tool_simple_constructs(self): + + with self.dataset_populator.test_history() as history_id: + + def _run_tool(tool_id, inputs): + self.dataset_populator.run_tool( + tool_id, + inputs, + history_id, + assert_ok=True, + ) + self.dataset_populator.wait_for_history(history_id) + + self._assert_file_counts(0, 0, 0, 0) + + hda1 = self.dataset_populator.new_dataset(history_id, content="1 2 3") + self.dataset_populator.wait_for_history(history_id) + hda1_input = {"src": "hda", "id": hda1["id"]} + + # One file uploaded, added to default object store ID. + self._assert_file_counts(1, 0, 0, 0) + + # should create two files in static object store. + _run_tool("multi_data_param", {"f1": hda1_input, "f2": hda1_input}) + self._assert_file_counts(1, 2, 0, 0) + + # should create two files in ebs object store. + create_10_inputs = { + "input1": hda1_input, + "input2": hda1_input, + } + _run_tool("create_10", create_10_inputs) + self._assert_file_counts(1, 2, 10, 0) + + # should create 10 files in S3 object store. + create_10_inputs = { + "__job_resource|__job_resource__select": "yes", + "__job_resource|how_store": "slow", + "input1": hda1_input, + "input2": hda1_input, + } + _run_tool("create_10", create_10_inputs) + self._assert_file_counts(1, 2, 10, 10) + + +def _files_count(directory): + return sum(len(files) for _, _, files in os.walk(directory))