From d39db9b6ea6be85ba46292faf05f02df3065331b Mon Sep 17 00:00:00 2001 From: Nicola Soranzo Date: Tue, 19 Nov 2024 01:11:01 +0000 Subject: [PATCH] Calculate hash for new non-deferred datasets when finishing a job This is configurable with two new options: - `calculate_dataset_hash`: in which cases Galaxy should calculate a hash for a new dataset. Possible values: 'always', 'upload' (the default), 'never'. - `hash_function`. Possible values: 'md5', 'sha1', 'sha256', 'sha512' Hashes are calculated via a Celery task, so currently only if the 'enable_celery_tasks' option is set to true. --- doc/source/admin/galaxy_options.rst | 27 ++++++++++ lib/galaxy/config/__init__.py | 9 ++++ lib/galaxy/config/sample/galaxy.yml.sample | 12 +++++ lib/galaxy/config/schemas/config_schema.yml | 22 +++++++++ lib/galaxy/jobs/__init__.py | 20 ++++++++ lib/galaxy/model/__init__.py | 2 +- lib/galaxy/model/deferred.py | 22 ++++----- lib/galaxy_test/api/test_datasets.py | 32 ++---------- lib/galaxy_test/base/populators.py | 7 +++ test/integration/test_dataset_hashing.py | 49 +++++++++++++++++++ ...test_materialize_dataset_instance_tasks.py | 21 ++++---- 11 files changed, 172 insertions(+), 51 deletions(-) create mode 100644 test/integration/test_dataset_hashing.py diff --git a/doc/source/admin/galaxy_options.rst b/doc/source/admin/galaxy_options.rst index a2a077224750..d7fe4cd5f41f 100644 --- a/doc/source/admin/galaxy_options.rst +++ b/doc/source/admin/galaxy_options.rst @@ -4719,6 +4719,33 @@ :Type: float +~~~~~~~~~~~~~~~~~~~~~~~~~~ +``calculate_dataset_hash`` +~~~~~~~~~~~~~~~~~~~~~~~~~~ + +:Description: + In which cases Galaxy should calculate a hash for a new dataset. + Dataset hashes can be used by the Galaxy job cache/search to check + if job inputs match. Setting the 'enable_celery_tasks' option to + true is also required for dataset hash calculation. Possible + values are: 'always', 'upload' (the default), 'never'. If set to + 'upload', the hash is calculated only for the outputs of upload + jobs. +:Default: ``upload`` +:Type: str + + +~~~~~~~~~~~~~~~~~ +``hash_function`` +~~~~~~~~~~~~~~~~~ + +:Description: + Hash function to use if 'calculate_dataset_hash' is enabled. + Possible values are: 'md5', 'sha1', 'sha256', 'sha512' +:Default: ``sha256`` +:Type: str + + ~~~~~~~~~~~~~~~~~~~~~ ``metadata_strategy`` ~~~~~~~~~~~~~~~~~~~~~ diff --git a/lib/galaxy/config/__init__.py b/lib/galaxy/config/__init__.py index d253b3d0d9b6..1dc273bc7240 100644 --- a/lib/galaxy/config/__init__.py +++ b/lib/galaxy/config/__init__.py @@ -46,6 +46,7 @@ from galaxy.util.custom_logging import LOGLV_TRACE from galaxy.util.dynamic import HasDynamicProperties from galaxy.util.facts import get_facts +from galaxy.util.hash_util import HashFunctionNameEnum from galaxy.util.properties import ( read_properties_from_file, running_from_source, @@ -716,6 +717,7 @@ class GalaxyAppConfiguration(BaseAppConfiguration, CommonConfigurationMixin): galaxy_data_manager_data_path: str galaxy_infrastructure_url: str hours_between_check: int + hash_function: HashFunctionNameEnum integrated_tool_panel_config: str involucro_path: str len_file_path: str @@ -897,6 +899,13 @@ def _process_config(self, kwargs: Dict[str, Any]) -> None: self.update_integrated_tool_panel = kwargs.get("update_integrated_tool_panel", True) self.galaxy_data_manager_data_path = self.galaxy_data_manager_data_path or self.tool_data_path self.tool_secret = kwargs.get("tool_secret", "") + if self.calculate_dataset_hash not in ("always", "upload", "never"): + raise ConfigurationError( + f"Unrecognized value for calculate_dataset_hash option: {self.calculate_dataset_hash}" + ) + if self.hash_function not in HashFunctionNameEnum.__members__: + raise ConfigurationError(f"Unrecognized value for hash_function option: {self.hash_function}") + self.hash_function = HashFunctionNameEnum[self.hash_function] self.metadata_strategy = kwargs.get("metadata_strategy", "directory") self.use_remote_user = self.use_remote_user or self.single_user self.fetch_url_allowlist_ips = parse_allowlist_ips(listify(kwargs.get("fetch_url_allowlist"))) diff --git a/lib/galaxy/config/sample/galaxy.yml.sample b/lib/galaxy/config/sample/galaxy.yml.sample index d2da6d1723ca..713b3788d26c 100644 --- a/lib/galaxy/config/sample/galaxy.yml.sample +++ b/lib/galaxy/config/sample/galaxy.yml.sample @@ -2545,6 +2545,18 @@ galaxy: # handler processes. Float values are allowed. #workflow_monitor_sleep: 1.0 + # In which cases Galaxy should calculate a hash for a new dataset. + # Dataset hashes can be used by the Galaxy job cache/search to check + # if job inputs match. Setting the 'enable_celery_tasks' option to + # true is also required for dataset hash calculation. Possible values + # are: 'always', 'upload' (the default), 'never'. If set to 'upload', + # the hash is calculated only for the outputs of upload jobs. + #calculate_dataset_hash: upload + + # Hash function to use if 'calculate_dataset_hash' is enabled. + # Possible values are: 'md5', 'sha1', 'sha256', 'sha512' + #hash_function: sha256 + # Determines how metadata will be set. Valid values are `directory`, # `extended`, `directory_celery` and `extended_celery`. In extended # mode jobs will decide if a tool run failed, the object stores diff --git a/lib/galaxy/config/schemas/config_schema.yml b/lib/galaxy/config/schemas/config_schema.yml index e659efd6d21c..d1e55b6a6eac 100644 --- a/lib/galaxy/config/schemas/config_schema.yml +++ b/lib/galaxy/config/schemas/config_schema.yml @@ -2788,6 +2788,7 @@ mapping: Avoiding making this a boolean because we may add options such as 'in-single-form-view' or 'in-simplified-workflow-views'. https://github.com/galaxyproject/galaxy/pull/9809/files#r461889109 + allow_user_dataset_purge: type: bool default: true @@ -3454,6 +3455,26 @@ mapping: decreased if extremely high job throughput is necessary, but doing so can increase CPU usage of handler processes. Float values are allowed. + calculate_dataset_hash: + type: str + default: upload + required: false + enum: ['always', 'upload', 'never'] + desc: | + In which cases Galaxy should calculate a hash for a new dataset. + Dataset hashes can be used by the Galaxy job cache/search to check if job inputs match. + Setting the 'enable_celery_tasks' option to true is also required for dataset hash calculation. + Possible values are: 'always', 'upload' (the default), 'never'. If set to 'upload', the + hash is calculated only for the outputs of upload jobs. + + hash_function: + type: str + default: sha256 + required: false + desc: | + Hash function to use if 'calculate_dataset_hash' is enabled. Possible values + are: 'md5', 'sha1', 'sha256', 'sha512' + metadata_strategy: type: str required: false @@ -3547,6 +3568,7 @@ mapping: default: always required: false reloadable: true + enum: ['always', 'onsuccess', 'never'] desc: | Clean up various bits of jobs left on the filesystem after completion. These bits include the job working directory, external metadata temporary files, diff --git a/lib/galaxy/jobs/__init__.py b/lib/galaxy/jobs/__init__.py index 0bf12219ce3f..22a909466b41 100644 --- a/lib/galaxy/jobs/__init__.py +++ b/lib/galaxy/jobs/__init__.py @@ -75,6 +75,7 @@ ObjectStorePopulator, serialize_static_object_store_config, ) +from galaxy.schema.tasks import ComputeDatasetHashTaskRequest from galaxy.structured_app import MinimalManagerApp from galaxy.tool_util.deps import requirements from galaxy.tool_util.output_checker import ( @@ -2038,6 +2039,25 @@ def fail(message=job.info, exception=None): dataset.full_delete() collected_bytes = 0 + # Calculate dataset hash + for dataset_assoc in output_dataset_associations: + dataset = dataset_assoc.dataset.dataset + if not dataset.purged and dataset.state != Dataset.states.DEFERRED and not dataset.hashes: + if self.app.config.calculate_dataset_hash == "always" or ( + self.app.config.calculate_dataset_hash == "upload" and job.tool_id in ("upload1", "__DATA_FETCH__") + ): + # Calculate dataset hash via a celery task + if self.app.config.enable_celery_tasks: + from galaxy.celery.tasks import compute_dataset_hash + + extra_files_path = dataset.extra_files_path if dataset.extra_files_path_exists() else None + request = ComputeDatasetHashTaskRequest( + dataset_id=dataset.id, + extra_files_path=extra_files_path, + hash_function=self.app.config.hash_function, + ) + compute_dataset_hash.delay(request=request) + user = job.user if user and collected_bytes > 0 and quota_source_info is not None and quota_source_info.use: user.adjust_total_disk_usage(collected_bytes, quota_source_info.label) diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py index f5f1085dd850..a5e03373dfb7 100644 --- a/lib/galaxy/model/__init__.py +++ b/lib/galaxy/model/__init__.py @@ -4290,7 +4290,7 @@ def get_file_name(self, sync_cache: bool = True) -> str: if not file_name and self.state not in (self.states.NEW, self.states.QUEUED): # Queued datasets can be assigned an object store and have a filename, but they aren't guaranteed to. # Anything after queued should have a file name. - log.warning(f"Failed to determine file name for dataset {self.id}") + log.warning(f"Failed to determine file name for dataset {self.id} in state {self.state}") return file_name else: filename = self.external_filename diff --git a/lib/galaxy/model/deferred.py b/lib/galaxy/model/deferred.py index 06eb995adc93..58f13f24a910 100644 --- a/lib/galaxy/model/deferred.py +++ b/lib/galaxy/model/deferred.py @@ -4,7 +4,6 @@ import shutil from typing import ( cast, - List, NamedTuple, Optional, Union, @@ -26,7 +25,6 @@ Dataset, DatasetCollection, DatasetCollectionElement, - DatasetHash, DatasetSource, DescribesHash, History, @@ -142,7 +140,7 @@ def ensure_materialized( sa_session.commit() object_store_populator.set_dataset_object_store_id(materialized_dataset) try: - path = self._stream_source(target_source, dataset_instance.datatype, materialized_dataset_hashes) + path = self._stream_source(target_source, dataset_instance.datatype, materialized_dataset) object_store.update_from_file(materialized_dataset, file_name=path) materialized_dataset.set_size() except Exception as e: @@ -152,9 +150,9 @@ def ensure_materialized( assert transient_path_mapper transient_paths = transient_path_mapper.transient_paths_for(dataset) # TODO: optimize this by streaming right to this path... - # TODO: take into acount transform and ensure we are and are not modifying the file as appropriate. + # TODO: take into account transform and ensure we are and are not modifying the file as appropriate. try: - path = self._stream_source(target_source, dataset_instance.datatype, materialized_dataset_hashes) + path = self._stream_source(target_source, dataset_instance.datatype, materialized_dataset) shutil.move(path, transient_paths.external_filename) materialized_dataset.external_filename = transient_paths.external_filename except Exception as e: @@ -178,9 +176,9 @@ def ensure_materialized( materialized_dataset_instance = cast(HistoryDatasetAssociation, dataset_instance) if exception_materializing is not None: materialized_dataset.state = Dataset.states.ERROR - materialized_dataset_instance.info = ( - f"Failed to materialize deferred dataset with exception: {exception_materializing}" - ) + error_msg = f"Failed to materialize deferred dataset with exception: {exception_materializing}" + materialized_dataset_instance.info = error_msg + log.error(error_msg) if attached: sa_session = self._sa_session if sa_session is None: @@ -206,7 +204,7 @@ def ensure_materialized( materialized_dataset_instance.metadata_deferred = False return materialized_dataset_instance - def _stream_source(self, target_source: DatasetSource, datatype, dataset_hashes: List[DatasetHash]) -> str: + def _stream_source(self, target_source: DatasetSource, datatype, dataset: Dataset) -> str: source_uri = target_source.source_uri if source_uri is None: raise Exception("Cannot stream from dataset source without specified source_uri") @@ -236,9 +234,11 @@ def _stream_source(self, target_source: DatasetSource, datatype, dataset_hashes: path = convert_result.converted_path if datatype_groom: datatype.groom_dataset_content(path) + # Grooming is not idempotent (e.g. paths in BAM headers), so invalidate hashes + dataset.hashes = [] - if dataset_hashes: - for dataset_hash in dataset_hashes: + if dataset.hashes: + for dataset_hash in dataset.hashes: _validate_hash(path, dataset_hash, "dataset contents") return path diff --git a/lib/galaxy_test/api/test_datasets.py b/lib/galaxy_test/api/test_datasets.py index 87ec35daf517..b74a230b213d 100644 --- a/lib/galaxy_test/api/test_datasets.py +++ b/lib/galaxy_test/api/test_datasets.py @@ -758,23 +758,13 @@ def test_composite_datatype_download(self, history_id): def test_compute_md5_on_primary_dataset(self, history_id): hda = self.dataset_populator.new_dataset(history_id, wait=True) - hda_details = self.dataset_populator.get_history_dataset_details(history_id, dataset=hda) - assert "hashes" in hda_details, str(hda_details.keys()) - hashes = hda_details["hashes"] - assert len(hashes) == 0 - self.dataset_populator.compute_hash(hda["id"]) hda_details = self.dataset_populator.get_history_dataset_details(history_id, dataset=hda) self.assert_hash_value(hda_details, "940cbe15c94d7e339dc15550f6bdcf4d", "MD5") def test_compute_sha1_on_composite_dataset(self, history_id): output = self.dataset_populator.fetch_hda(history_id, COMPOSITE_DATA_FETCH_REQUEST_1, wait=True) - hda_details = self.dataset_populator.get_history_dataset_details(history_id, dataset=output) - assert "hashes" in hda_details, str(hda_details.keys()) - hashes = hda_details["hashes"] - assert len(hashes) == 0 - - self.dataset_populator.compute_hash(hda_details["id"], hash_function="SHA-256", extra_files_path="Roadmaps") + self.dataset_populator.compute_hash(output["id"], hash_function="SHA-256", extra_files_path="Roadmaps") hda_details = self.dataset_populator.get_history_dataset_details(history_id, dataset=output) self.assert_hash_value( hda_details, @@ -785,11 +775,6 @@ def test_compute_sha1_on_composite_dataset(self, history_id): def test_duplicated_hash_requests_on_primary(self, history_id): hda = self.dataset_populator.new_dataset(history_id, wait=True) - hda_details = self.dataset_populator.get_history_dataset_details(history_id, dataset=hda) - assert "hashes" in hda_details, str(hda_details.keys()) - hashes = hda_details["hashes"] - assert len(hashes) == 0 - self.dataset_populator.compute_hash(hda["id"]) self.dataset_populator.compute_hash(hda["id"]) hda_details = self.dataset_populator.get_history_dataset_details(history_id, dataset=hda) @@ -797,19 +782,12 @@ def test_duplicated_hash_requests_on_primary(self, history_id): def test_duplicated_hash_requests_on_extra_files(self, history_id): output = self.dataset_populator.fetch_hda(history_id, COMPOSITE_DATA_FETCH_REQUEST_1, wait=True) - hda_details = self.dataset_populator.get_history_dataset_details(history_id, dataset=output) - assert "hashes" in hda_details, str(hda_details.keys()) - hashes = hda_details["hashes"] - assert len(hashes) == 0 - # 4 unique requests, but make them twice... for _ in range(2): - self.dataset_populator.compute_hash(hda_details["id"], hash_function="SHA-256", extra_files_path="Roadmaps") - self.dataset_populator.compute_hash(hda_details["id"], hash_function="SHA-1", extra_files_path="Roadmaps") - self.dataset_populator.compute_hash(hda_details["id"], hash_function="MD5", extra_files_path="Roadmaps") - self.dataset_populator.compute_hash( - hda_details["id"], hash_function="SHA-256", extra_files_path="Sequences" - ) + self.dataset_populator.compute_hash(output["id"], hash_function="SHA-256", extra_files_path="Roadmaps") + self.dataset_populator.compute_hash(output["id"], hash_function="SHA-1", extra_files_path="Roadmaps") + self.dataset_populator.compute_hash(output["id"], hash_function="MD5", extra_files_path="Roadmaps") + self.dataset_populator.compute_hash(output["id"], hash_function="SHA-256", extra_files_path="Sequences") hda_details = self.dataset_populator.get_history_dataset_details(history_id, dataset=output) self.assert_hash_value(hda_details, "ce0c0ef1073317ff96c896c249b002dc", "MD5", extra_files_path="Roadmaps") diff --git a/lib/galaxy_test/base/populators.py b/lib/galaxy_test/base/populators.py index 684689de2f6f..3e47c70c18f8 100644 --- a/lib/galaxy_test/base/populators.py +++ b/lib/galaxy_test/base/populators.py @@ -1398,6 +1398,13 @@ def validated(): return wait_on(validated, "dataset validation") + def wait_for_dataset_hashes(self, history_id: str, dataset_id: str): + def dataset_hashes_present(): + hda = self.get_history_dataset_details(history_id=history_id, dataset_id=dataset_id) + return hda["hashes"] or None + + return wait_on(dataset_hashes_present, "dataset hash presence") + def setup_history_for_export_testing(self, history_name): using_requirement("new_history") history_id = self.new_history(name=history_name) diff --git a/test/integration/test_dataset_hashing.py b/test/integration/test_dataset_hashing.py new file mode 100644 index 000000000000..a3a123b735f1 --- /dev/null +++ b/test/integration/test_dataset_hashing.py @@ -0,0 +1,49 @@ +from typing import Optional + +from galaxy_test.base.populators import DatasetPopulator +from galaxy_test.driver import integration_util + + +class TestDatasetHashingIntegration(integration_util.IntegrationTestCase): + dataset_populator: DatasetPopulator + calculate_dataset_hash: Optional[str] = None + + def setUp(self) -> None: + super().setUp() + self.dataset_populator = DatasetPopulator(self.galaxy_interactor) + + @classmethod + def handle_galaxy_config_kwds(cls, config) -> None: + super().handle_galaxy_config_kwds(config) + if cls.calculate_dataset_hash is not None: + config["enable_celery_tasks"] = True + config["calculate_dataset_hash"] = cls.calculate_dataset_hash + + def test_hashing(self, history_id: str) -> None: + hda = self.dataset_populator.new_dataset(history_id, wait=True) + if self.calculate_dataset_hash in [None, "always", "upload"]: + hashes = self.dataset_populator.wait_for_dataset_hashes(history_id=history_id, dataset_id=hda["id"]) + assert hashes[0]["hash_value"] == "a17dcdfd36f47303a4824f1309d43ac14d7491ab3b8abb28782ac8e8d3b680ea" + else: + assert hda["hashes"] == [], hda + inputs = {"input1": {"src": "hda", "id": hda["id"]}} + run_response = self.dataset_populator.run_tool_raw("cat1", inputs=inputs, history_id=history_id) + self.dataset_populator.wait_for_tool_run(history_id=history_id, run_response=run_response) + cat_dataset = self.dataset_populator.get_history_dataset_details(history_id=history_id) + if self.calculate_dataset_hash == "always": + hashes = self.dataset_populator.wait_for_dataset_hashes(history_id=history_id, dataset_id=cat_dataset["id"]) + assert hashes[0]["hash_value"] == "a17dcdfd36f47303a4824f1309d43ac14d7491ab3b8abb28782ac8e8d3b680ea" + else: + assert cat_dataset["hashes"] == [], cat_dataset + + +class TestDatasetHashingAlwaysIntegration(TestDatasetHashingIntegration): + calculate_dataset_hash = "always" + + +class TestDatasetHashingUploadIntegration(TestDatasetHashingIntegration): + calculate_dataset_hash = "upload" + + +class TestDatasetHashingNeverIntegration(TestDatasetHashingIntegration): + calculate_dataset_hash = "never" diff --git a/test/integration/test_materialize_dataset_instance_tasks.py b/test/integration/test_materialize_dataset_instance_tasks.py index 817098994a5f..aef30d0a29c6 100644 --- a/test/integration/test_materialize_dataset_instance_tasks.py +++ b/test/integration/test_materialize_dataset_instance_tasks.py @@ -160,7 +160,7 @@ def test_upload_vs_materialize_simplest_upload(self, history_id: str): assert len(uploaded_details["sources"]) == 1 content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output, assert_ok=True) assert content == "This is a line of text." - new_history_id = self._reupload_and_then_materialize(history_id, output) + new_history_id = self._reupload_and_then_materialize(output) content = self.dataset_populator.get_history_dataset_content(new_history_id, hid=2, assert_ok=False) assert content == "This is a line of text." @@ -186,7 +186,7 @@ def test_upload_vs_materialize_to_posix_lines(self, history_id: str): assert len(transform) == 1 content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output, assert_ok=True) assert content == "This is a line of text.\n" - new_history_id = self._reupload_and_then_materialize(history_id, output) + new_history_id = self._reupload_and_then_materialize(output) content = self.dataset_populator.get_history_dataset_content(new_history_id, hid=2, assert_ok=False) assert content == "This is a line of text.\n" @@ -212,7 +212,7 @@ def test_upload_vs_materialize_space_to_tab(self, history_id: str): assert len(transform) == 1 content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output, assert_ok=True) assert content == "This\tis\ta\tline\tof\ttext." - new_history_id = self._reupload_and_then_materialize(history_id, output) + new_history_id = self._reupload_and_then_materialize(output) content = self.dataset_populator.get_history_dataset_content(new_history_id, hid=2, assert_ok=False) assert content == "This\tis\ta\tline\tof\ttext." @@ -239,7 +239,7 @@ def test_upload_vs_materialize_to_posix_and_space_to_tab(self, history_id: str): assert len(transform) == 2 content = self.dataset_populator.get_history_dataset_content(history_id, dataset=output, assert_ok=True) assert content == "This\tis\ta\tline\tof\ttext.\n" - new_history_id = self._reupload_and_then_materialize(history_id, output) + new_history_id = self._reupload_and_then_materialize(output) content = self.dataset_populator.get_history_dataset_content(new_history_id, hid=2, assert_ok=False) assert content == "This\tis\ta\tline\tof\ttext.\n" @@ -262,20 +262,17 @@ def test_upload_vs_materialize_grooming(self, history_id: str): transform = source_0["transform"] assert isinstance(transform, list) assert len(transform) == 1 - original_details = self.dataset_populator.get_history_dataset_details( - history_id, dataset=output, assert_ok=True - ) - new_history_id = self._reupload_and_then_materialize(history_id, output) + new_history_id = self._reupload_and_then_materialize(output) new_details = self.dataset_populator.get_history_dataset_details(new_history_id, hid=2, assert_ok=False) - for key in original_details.keys(): + for key in uploaded_details.keys(): if key in ["metadata_bam_header", "metadata_bam_index"]: # differs because command-line different, index path different, and such... continue if key.startswith("metadata_"): - assert original_details[key] == new_details[key], f"Mismatched on key {key}" - assert original_details["file_ext"] == new_details["file_ext"] + assert uploaded_details[key] == new_details[key], f"Mismatched on key {key}" + assert uploaded_details["file_ext"] == new_details["file_ext"] - def _reupload_and_then_materialize(self, history_id, dataset): + def _reupload_and_then_materialize(self, dataset): new_history_id, uploaded_hdas = self.dataset_populator.reupload_contents(dataset) assert len(uploaded_hdas) == 1 deferred_hda = uploaded_hdas[0]