diff --git a/doc/source/admin/galaxy_options.rst b/doc/source/admin/galaxy_options.rst index 4852c9e9456..821200d0bde 100644 --- a/doc/source/admin/galaxy_options.rst +++ b/doc/source/admin/galaxy_options.rst @@ -3586,6 +3586,21 @@ :Type: int +~~~~~~~~~~~~~~~~~~~~~~~~ +``max_discovered_files`` +~~~~~~~~~~~~~~~~~~~~~~~~ + +:Description: + Set this to a positive integer value to limit the number of + datasets that can be discovered by a single job. This prevents + accidentally creating large numbers of datasets when running tools + that create a potentially unlimited number of output datasets, + such as tools that split a file into a collection of datasets for + each line in an input dataset. +:Default: ``10000`` +:Type: int + + ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ ``history_local_serial_workflow_scheduling`` ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ diff --git a/lib/galaxy/app_unittest_utils/galaxy_mock.py b/lib/galaxy/app_unittest_utils/galaxy_mock.py index 1fb9dc00601..192e144079e 100644 --- a/lib/galaxy/app_unittest_utils/galaxy_mock.py +++ b/lib/galaxy/app_unittest_utils/galaxy_mock.py @@ -189,6 +189,7 @@ class MockAppConfig(GalaxyDataTestConfig, CommonConfigurationMixin): self.monitor_thread_join_timeout = 1 self.integrated_tool_panel_config = None self.vault_config_file = kwargs.get('vault_config_file') + self.max_discovered_files = 10000 @property def config_dict(self): diff --git a/lib/galaxy/config/sample/galaxy.yml.sample b/lib/galaxy/config/sample/galaxy.yml.sample index 71f5500e749..67604bd99ba 100644 --- a/lib/galaxy/config/sample/galaxy.yml.sample +++ b/lib/galaxy/config/sample/galaxy.yml.sample @@ -1759,6 +1759,14 @@ galaxy: # creating datasets in batches. #flush_per_n_datasets: 1000 + # Set this to a positive integer value to limit the number of datasets + # that can be discovered by a single job. This prevents accidentally + # creating large numbers of datasets when running tools that create a + # potentially unlimited number of output datasets, such as tools that + # split a file into a collection of datasets for each line in an input + # dataset. + #max_discovered_files: 10000 + # Force serial scheduling of workflows within the context of a # particular history #history_local_serial_workflow_scheduling: false diff --git a/lib/galaxy/job_execution/output_collect.py b/lib/galaxy/job_execution/output_collect.py index c7df1b15a2f..05dbc41b29a 100644 --- a/lib/galaxy/job_execution/output_collect.py +++ b/lib/galaxy/job_execution/output_collect.py @@ -5,11 +5,17 @@ import operator import os import re from tempfile import NamedTemporaryFile -from typing import Optional +from typing import Callable, Dict, List, Optional, Union from sqlalchemy.orm.scoping import ScopedSession -import galaxy.model +from galaxy.model import ( + HistoryDatasetAssociation, + HistoryDatasetCollectionAssociation, + Job, + JobToOutputDatasetAssociation, + LibraryDatasetDatasetAssociation, +) from galaxy.model.dataset_collections import builder from galaxy.model.dataset_collections.structure import UninitializedTree from galaxy.model.dataset_collections.type_description import COLLECTION_TYPE_DESCRIPTION_FACTORY @@ -38,7 +44,8 @@ from galaxy.tool_util.parser.output_objects import ( ToolOutputCollection, ) from galaxy.util import ( - unicodify + shrink_and_unicodify, + unicodify, ) DATASET_ID_TOKEN = "DATASET_ID" @@ -176,12 +183,15 @@ def collect_dynamic_outputs( class BaseJobContext: + max_discovered_files: Union[int, float] + def add_dataset_collection(self, collection): pass def find_files(self, output_name, collection, dataset_collectors): filenames = {} for discovered_file in discover_files(output_name, self.tool_provided_metadata, dataset_collectors, self.job_working_directory, collection): + self.increment_discovered_file_count() filenames[discovered_file.path] = discovered_file return filenames @@ -202,7 +212,9 @@ class JobContext(ModelPersistenceContext, BaseJobContext): input_dbkey, object_store, final_job_state, - flush_per_n_datasets=None): + max_discovered_files: Optional[int], + flush_per_n_datasets=None, + ): self.tool = tool self._metadata_source_provider = metadata_source_provider self._permission_provider = permission_provider @@ -215,6 +227,8 @@ class JobContext(ModelPersistenceContext, BaseJobContext): self._object_store = object_store self.final_job_state = final_job_state self._flush_per_n_datasets = flush_per_n_datasets + self.max_discovered_files = float('inf') if max_discovered_files is None else max_discovered_files + self.discovered_file_count = 0 self._tag_handler = None @property @@ -241,7 +255,7 @@ class JobContext(ModelPersistenceContext, BaseJobContext): return self._metadata_source_provider @property - def job(self) -> galaxy.model.Job: + def job(self) -> Job: return self._job @property @@ -277,7 +291,7 @@ class JobContext(ModelPersistenceContext, BaseJobContext): return library_folder def get_hdca(self, object_id): - hdca = self.sa_session.query(galaxy.model.HistoryDatasetCollectionAssociation).get(int(object_id)) + hdca = self.sa_session.query(HistoryDatasetCollectionAssociation).get(int(object_id)) return hdca def create_library_folder(self, parent_folder, name, description): @@ -296,7 +310,7 @@ class JobContext(ModelPersistenceContext, BaseJobContext): return hdca def add_output_dataset_association(self, name, dataset): - assoc = galaxy.model.JobToOutputDatasetAssociation(name, dataset) + assoc = JobToOutputDatasetAssociation(name, dataset) assoc.job = self.job self.sa_session.add(assoc) @@ -365,13 +379,15 @@ class JobContext(ModelPersistenceContext, BaseJobContext): class SessionlessJobContext(SessionlessModelPersistenceContext, BaseJobContext): - def __init__(self, metadata_params, tool_provided_metadata, object_store, export_store, import_store, working_directory, final_job_state): + def __init__(self, metadata_params, tool_provided_metadata, object_store, export_store, import_store, working_directory, final_job_state, max_discovered_files: Optional[int]): # TODO: use a metadata source provider... (pop from inputs and add parameter) super().__init__(object_store, export_store, working_directory) self.metadata_params = metadata_params self.tool_provided_metadata = tool_provided_metadata self.import_store = import_store self.final_job_state = final_job_state + self.max_discovered_files = float('inf') if max_discovered_files is None else max_discovered_files + self.discovered_file_count = 0 def output_collection_def(self, name): tool_as_dict = self.metadata_params["tool"] @@ -397,7 +413,7 @@ class SessionlessJobContext(SessionlessModelPersistenceContext, BaseJobContext): return "non-session bound job" def get_hdca(self, object_id): - hdca = self.import_store.sa_session.query(galaxy.model.HistoryDatasetCollectionAssociation).find(int(object_id)) + hdca = self.import_store.sa_session.query(HistoryDatasetCollectionAssociation).find(int(object_id)) if hdca: self.export_store.add_dataset_collection(hdca) for collection_dataset in hdca.dataset_instances: @@ -424,15 +440,15 @@ class SessionlessJobContext(SessionlessModelPersistenceContext, BaseJobContext): return self.metadata_params.get("implicit_collection_jobs_association_id") -def collect_primary_datasets(job_context, output, input_ext): +def collect_primary_datasets(job_context: Union[JobContext, SessionlessJobContext], output, input_ext): job_working_directory = job_context.job_working_directory # Loop through output file names, looking for generated primary # datasets in form specified by discover dataset patterns or in tool provided metadata. primary_output_assigned = False new_outdata_name = None - primary_datasets = {} - storage_callbacks = [] + primary_datasets: Dict[str, Dict[str, Union[HistoryDatasetAssociation, LibraryDatasetDatasetAssociation]]] = {} + storage_callbacks: List[Callable] = [] for output_index, (name, outdata) in enumerate(output.items()): dataset_collectors = [DEFAULT_DATASET_COLLECTOR] output_def = job_context.output_def(name) @@ -440,6 +456,7 @@ def collect_primary_datasets(job_context, output, input_ext): dataset_collectors = [dataset_collector(description) for description in output_def.dataset_collector_descriptions] filenames = {} for discovered_file in discover_files(name, job_context.tool_provided_metadata, dataset_collectors, job_working_directory, outdata): + job_context.increment_discovered_file_count() filenames[discovered_file.path] = discovered_file for filename_index, (filename, discovered_file) in enumerate(filenames.items()): extra_file_collector = discovered_file.collector @@ -704,6 +721,6 @@ def collect_extra_files(object_store, dataset, job_working_directory): def collect_shrinked_content_from_path(path): try: with open(path, 'rb') as fh: - return galaxy.util.shrink_and_unicodify(fh.read().strip()) + return shrink_and_unicodify(fh.read().strip()) except FileNotFoundError: return None diff --git a/lib/galaxy/jobs/__init__.py b/lib/galaxy/jobs/__init__.py index b908a35889a..4c47dadea16 100644 --- a/lib/galaxy/jobs/__init__.py +++ b/lib/galaxy/jobs/__init__.py @@ -51,6 +51,7 @@ from galaxy.jobs.mapper import ( from galaxy.jobs.runners import BaseJobRunner, JobState from galaxy.metadata import get_metadata_compute_strategy from galaxy.model import store +from galaxy.model.store.discover import MaxDiscoveredFilesExceededError from galaxy.objectstore import ObjectStorePopulator from galaxy.structured_app import MinimalManagerApp from galaxy.tool_util.deps import requirements @@ -1731,6 +1732,8 @@ class JobWrapper(HasResourceParameters): tag_handler=self.app.tag_handler.create_tag_handler_session(), ) import_model_store.perform_import(history=job.history, job=job) + if job.state == job.states.ERROR: + final_job_state = job.state except store.FileTracebackException as e: job.traceback = e.traceback log.exception(f"Problem generating command line for Job {job.id}.\n{job.traceback}") @@ -1762,7 +1765,11 @@ class JobWrapper(HasResourceParameters): ) if not final_job_state == job.states.ERROR: dataset_assoc.dataset.dataset.state = model.Dataset.states.OK - self.discover_outputs(job, inp_data, out_data, out_collections, final_job_state=final_job_state) + try: + self.discover_outputs(job, inp_data, out_data, out_collections, final_job_state=final_job_state) + except MaxDiscoveredFilesExceededError as e: + final_job_state = job.states.ERROR + job.job_messages = [str(e)] if job.states.ERROR == final_job_state: for dataset_assoc in output_dataset_associations: @@ -2061,6 +2068,7 @@ class JobWrapper(HasResourceParameters): tool=self.tool, job=job, max_metadata_value_size=self.app.config.max_metadata_value_size, + max_discovered_files=self.app.config.max_discovered_files, validate_outputs=self.validate_outputs, link_data_only=self.__link_file_check(), **kwds) diff --git a/lib/galaxy/metadata/__init__.py b/lib/galaxy/metadata/__init__.py index eaf3ad18438..89d3185d9ed 100644 --- a/lib/galaxy/metadata/__init__.py +++ b/lib/galaxy/metadata/__init__.py @@ -47,7 +47,7 @@ class MetadataCollectionStrategy(metaclass=abc.ABCMeta): output_fnames=None, config_root=None, use_bin=False, config_file=None, datatypes_config=None, job_metadata=None, provided_metadata_style=None, compute_tmp_dir=None, - include_command=True, max_metadata_value_size=0, + include_command=True, max_metadata_value_size=0, max_discovered_files=None, object_store_conf=None, tool=None, job=None, kwds=None): """Setup files needed for external metadata collection. @@ -106,7 +106,7 @@ class PortableDirectoryMetadataGenerator(MetadataCollectionStrategy): output_fnames=None, config_root=None, use_bin=False, config_file=None, datatypes_config=None, job_metadata=None, provided_metadata_style=None, compute_tmp_dir=None, - include_command=True, max_metadata_value_size=0, + include_command=True, max_metadata_value_size=0, max_discovered_files=None, validate_outputs=False, object_store_conf=None, tool=None, job=None, link_data_only=False, kwds=None): @@ -152,6 +152,7 @@ class PortableDirectoryMetadataGenerator(MetadataCollectionStrategy): "provided_metadata_style": provided_metadata_style, "datatypes_config": datatypes_config, "max_metadata_value_size": max_metadata_value_size, + "max_discovered_files": max_discovered_files, "outputs": outputs, } diff --git a/lib/galaxy/metadata/set_metadata.py b/lib/galaxy/metadata/set_metadata.py index d839c640e55..90755ce774c 100644 --- a/lib/galaxy/metadata/set_metadata.py +++ b/lib/galaxy/metadata/set_metadata.py @@ -46,6 +46,7 @@ from galaxy.model import ( ) from galaxy.model.custom_types import total_size from galaxy.model.metadata import MetadataTempFile +from galaxy.model.store.discover import MaxDiscoveredFilesExceededError from galaxy.objectstore import build_object_store_from_config from galaxy.tool_util.output_checker import ( check_output, @@ -145,6 +146,7 @@ def set_metadata_portable(): job_metadata = metadata_params["job_metadata"] provided_metadata_style = metadata_params.get("provided_metadata_style") max_metadata_value_size = metadata_params.get("max_metadata_value_size") or 0 + max_discovered_files = metadata_params.get("max_discovered_files") outputs = metadata_params["outputs"] datatypes_registry = validate_and_load_datatypes_config(datatypes_config) @@ -163,6 +165,7 @@ def set_metadata_portable(): export_store = None final_job_state = Job.states.OK + job_messages = [] if extended_metadata_collection: tool_dict = metadata_params["tool"] stdio_exit_code_dicts, stdio_regex_dicts = tool_dict["stdio_exit_codes"], tool_dict["stdio_regexes"] @@ -234,11 +237,9 @@ def set_metadata_portable(): import_model_store = None tool_script_file = os.path.join(tool_job_working_directory, 'tool_script.sh') - if import_model_store and export_store and os.path.exists(tool_script_file): + job = None + if import_model_store and export_store: job = next(iter(import_model_store.sa_session.objects[Job].values())) - with open(tool_script_file) as command_fh: - job.command_line = command_fh.read().strip() - export_store.export_job(job, include_job_data=False) job_context = SessionlessJobContext( metadata_params, @@ -248,8 +249,41 @@ def set_metadata_portable(): import_model_store, os.path.join(tool_job_working_directory, "working"), final_job_state=final_job_state, + max_discovered_files=max_discovered_files, ) + if extended_metadata_collection: + # discover extra outputs... + output_collections = {} + for name, output_collection in metadata_params["output_collections"].items(): + # TODO: remove HistoryDatasetCollectionAssociation fallback on 22.01, model_class used to not be serialized prior to 21.09 + model_class = output_collection.get('model_class', 'HistoryDatasetCollectionAssociation') + collection = import_model_store.sa_session.query(getattr(galaxy.model, model_class)).find(output_collection["id"]) + output_collections[name] = collection + output_instances = {} + for name, output in metadata_params["outputs"].items(): + klass = getattr(galaxy.model, output.get('model_class', 'HistoryDatasetAssociation')) + output_instances[name] = import_model_store.sa_session.query(klass).find(output["id"]) + + input_ext = json.loads(metadata_params["job_params"].get("__input_ext") or '"data"') + try: + collect_primary_datasets( + job_context, + output_instances, + input_ext=input_ext, + ) + collect_dynamic_outputs(job_context, output_collections) + except MaxDiscoveredFilesExceededError as e: + final_job_state = Job.states.ERROR + job_messages.append(str(e)) + if job: + job.job_messages = job_messages + job.state = final_job_state + if os.path.exists(tool_script_file): + with open(tool_script_file) as command_fh: + job.command_line = command_fh.read().strip() + export_store.export_job(job, include_job_data=False) + unnamed_id_to_path = {} for unnamed_output_dict in job_context.tool_provided_metadata.get_unnamed_outputs(): destination = unnamed_output_dict["destination"] @@ -370,27 +404,6 @@ def set_metadata_portable(): except Exception: json.dump((False, traceback.format_exc()), open(filename_results_code, 'wt+')) # setting metadata has failed somehow - if extended_metadata_collection: - # discover extra outputs... - output_collections = {} - for name, output_collection in metadata_params["output_collections"].items(): - # TODO: remove HistoryDatasetCollectionAssociation fallback on 22.01, model_class used to not be serialized prior to 21.09 - model_class = output_collection.get('model_class', 'HistoryDatasetCollectionAssociation') - collection = import_model_store.sa_session.query(getattr(galaxy.model, model_class)).find(output_collection["id"]) - output_collections[name] = collection - outputs = {} - for name, output in metadata_params["outputs"].items(): - klass = getattr(galaxy.model, output.get('model_class', 'HistoryDatasetAssociation')) - outputs[name] = import_model_store.sa_session.query(klass).find(output["id"]) - - input_ext = json.loads(metadata_params["job_params"].get("__input_ext") or '"data"') - collect_primary_datasets( - job_context, - outputs, - input_ext=input_ext, - ) - collect_dynamic_outputs(job_context, output_collections) - if export_store: export_store._finalize() write_job_metadata(tool_job_working_directory, job_metadata, set_meta, tool_provided_metadata) diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py index 500e2d80fd3..6e4e70452c8 100644 --- a/lib/galaxy/model/__init__.py +++ b/lib/galaxy/model/__init__.py @@ -1358,6 +1358,7 @@ class Job(Base, JobLike, UsesCreateAndUpdateTime, Dictifiable, Serializable): job_attrs['exit_code'] = self.exit_code job_attrs['create_time'] = self.create_time.isoformat() job_attrs['update_time'] = self.update_time.isoformat() + job_attrs['job_messages'] = self.job_messages # Get the job's parameters param_dict = self.raw_param_dict() diff --git a/lib/galaxy/model/store/__init__.py b/lib/galaxy/model/store/__init__.py index 724bea5b2a1..0153b1f7ca1 100644 --- a/lib/galaxy/model/store/__init__.py +++ b/lib/galaxy/model/store/__init__.py @@ -705,6 +705,7 @@ class ModelImportStore(metaclass=abc.ABCMeta): assert self.import_options.allow_edit job = self.sa_session.query(model.Job).get(job_attrs["id"]) self._connect_job_io(job, job_attrs, _find_hda, _find_hdca, _find_dce) + self._set_job_attributes(job, job_attrs, force_terminal=False) # Don't edit job continue @@ -715,25 +716,8 @@ class ModelImportStore(metaclass=abc.ABCMeta): imported_job.imported = True imported_job.tool_id = job_attrs['tool_id'] imported_job.tool_version = job_attrs['tool_version'] - raw_state = job_attrs['state'] - if raw_state not in model.Job.terminal_states: - raw_state = model.Job.states.ERROR - imported_job.set_state(raw_state) - imported_job.info = job_attrs.get('info', None) - imported_job.exit_code = job_attrs.get('exit_code', None) - imported_job.traceback = job_attrs.get('traceback', None) - if 'stdout' in job_attrs: - # Pre 19.05 export. - imported_job.tool_stdout = job_attrs.get('stdout', None) - imported_job.tool_stderr = job_attrs.get('stderr', None) - else: - # Post 19.05 export with separated I/O - imported_job.tool_stdout = job_attrs.get('tool_stdout', None) - imported_job.job_stdout = job_attrs.get('job_stdout', None) - imported_job.tool_stderr = job_attrs.get('tool_stderr', None) - imported_job.job_stderr = job_attrs.get('job_stderr', None) + self._set_job_attributes(imported_job, job_attrs, force_terminal=True) - imported_job.command_line = job_attrs.get('command_line') try: imported_job.create_time = datetime.datetime.strptime(job_attrs["create_time"], "%Y-%m-%dT%H:%M:%S.%f") imported_job.update_time = datetime.datetime.strptime(job_attrs["update_time"], "%Y-%m-%dT%H:%M:%S.%f") @@ -936,6 +920,29 @@ class BaseDirectoryImportModelStore(ModelImportStore): except FileNotFoundError: return [] + def _set_job_attributes(self, imported_job, job_attrs, force_terminal=False): + ATTRIBUTES = ( + 'info', + 'exit_code', + 'traceback', + 'job_messages', + 'tool_stdout', + 'tool_stderr', + 'job_stdout', + 'job_stderr' + ) + for attribute in ATTRIBUTES: + value = job_attrs.get(attribute) + if value is not None: + setattr(imported_job, attribute, value) + if 'stdout' in job_attrs: + imported_job.tool_stdout = job_attrs.get('stdout') + imported_job.tool_stderr = job_attrs.get('stderr') + raw_state = job_attrs.get('state') + if force_terminal and raw_state and raw_state not in model.Job.terminal_states: + raw_state = model.Job.states.ERROR + imported_job.set_state(raw_state) + class DirectoryImportModelStore1901(BaseDirectoryImportModelStore): object_key = 'hid' @@ -1245,14 +1252,14 @@ class DirectoryModelExportStore(ModelExportStore): def __enter__(self): return self - def export_job(self, job, tool=None, include_job_data=True): + def export_job(self, job: model.Job, tool=None, include_job_data=True): self.export_jobs([job], include_job_data=include_job_data) tool_source = getattr(tool, 'tool_source', None) if tool_source: with open(os.path.join(self.export_directory, 'tool.xml'), 'w') as out: out.write(tool_source.to_string()) - def export_jobs(self, jobs, jobs_attrs=None, include_job_data=True): + def export_jobs(self, jobs: List[model.Job], jobs_attrs=None, include_job_data=True): """ Export jobs. @@ -1267,12 +1274,12 @@ class DirectoryModelExportStore(ModelExportStore): if include_job_data: # -- Get input, output datasets. -- - input_dataset_mapping = {} - output_dataset_mapping = {} - input_dataset_collection_mapping = {} - input_dataset_collection_element_mapping = {} - output_dataset_collection_mapping = {} - implicit_output_dataset_collection_mapping = {} + input_dataset_mapping: Dict[str, List[model.DatasetInstance]] = {} + output_dataset_mapping: Dict[str, List[model.DatasetInstance]] = {} + input_dataset_collection_mapping: Dict[str, List[model.DatasetCollectionInstance]] = {} + input_dataset_collection_element_mapping: Dict[str, List[model.DatasetCollectionElement]] = {} + output_dataset_collection_mapping: Dict[str, List[model.DatasetCollectionInstance]] = {} + implicit_output_dataset_collection_mapping: Dict[str, List[model.DatasetCollection]] = {} for assoc in job.input_datasets: # Optional data inputs will not have a dataset. diff --git a/lib/galaxy/model/store/discover.py b/lib/galaxy/model/store/discover.py index 4928295f946..f4da97d2509 100644 --- a/lib/galaxy/model/store/discover.py +++ b/lib/galaxy/model/store/discover.py @@ -11,7 +11,16 @@ import os from collections import ( namedtuple, ) -from typing import Any, Callable, Dict, List, NamedTuple, Optional +from typing import ( + Any, + Callable, + Dict, + List, + NamedTuple, + Optional, + TYPE_CHECKING, + Union, +) from sqlalchemy.orm.scoping import ScopedSession @@ -29,18 +38,30 @@ from galaxy.util import ( ) from galaxy.util.hash_util import HASH_NAME_MAP +if TYPE_CHECKING: + from galaxy.job_execution.output_collect import JobContext, SessionlessJobContext + + log = logging.getLogger(__name__) UNSET = object() DEFAULT_CHUNK_SIZE = 1000 +class MaxDiscoveredFilesExceededError(ValueError): + pass + + class ModelPersistenceContext(metaclass=abc.ABCMeta): """Class for creating datasets while finding files. This class implement the create_dataset method that takes care of populating metadata required for datasets and other potential model objects. """ + + max_discovered_files = float('inf') + discovered_file_count: int + def create_dataset( self, ext, @@ -400,6 +421,11 @@ class ModelPersistenceContext(metaclass=abc.ABCMeta): def flush(self): """If database bound, flush the persisted objects to ensure IDs.""" + def increment_discovered_file_count(self): + self.discovered_file_count += 1 + if self.discovered_file_count > self.max_discovered_files: + raise MaxDiscoveredFilesExceededError(f"Job generated more than maximum number ({self.max_discovered_files}) of output datasets") + class PermissionProvider(metaclass=abc.ABCMeta): """Interface for working with permissions while importing datasets with ModelPersistenceContext.""" @@ -457,6 +483,8 @@ class SessionlessModelPersistenceContext(ModelPersistenceContext): self._object_store = object_store self.export_store = export_store self._flush_per_n_datasets = None + self.discovered_file_count = 0 + self.max_discovered_files = float('inf') self.job_working_directory = working_directory # TODO: rename... @@ -607,7 +635,7 @@ def persist_target_to_export_store(target_dict, export_store, object_store, work export_store.add_dataset_collection(hdca) -def persist_elements_to_hdca(model_persistence_context, elements, hdca, collector=None): +def persist_elements_to_hdca(model_persistence_context: Union['JobContext', 'SessionlessJobContext', SessionlessModelPersistenceContext], elements, hdca, collector=None): filenames = {} def add_to_discovered_files(elements, parent_identifiers=None): @@ -616,7 +644,7 @@ def persist_elements_to_hdca(model_persistence_context, elements, hdca, collecto if "elements" in element: add_to_discovered_files(element["elements"], parent_identifiers + [element["name"]]) else: - discovered_file = discovered_file_for_element(element, model_persistence_context.job_working_directory, parent_identifiers, collector=collector) + discovered_file = discovered_file_for_element(element, model_persistence_context, parent_identifiers, collector=collector) filenames[discovered_file.path] = discovered_file add_to_discovered_files(elements) @@ -640,7 +668,7 @@ def persist_elements_to_folder(model_persistence_context, elements, library_fold nested_folder = model_persistence_context.create_library_folder(library_folder, name, description) persist_elements_to_folder(model_persistence_context, element["elements"], nested_folder) else: - discovered_file = discovered_file_for_element(element, model_persistence_context.job_working_directory) + discovered_file = discovered_file_for_element(element, model_persistence_context) fields_match = discovered_file.match designation = fields_match.designation visible = fields_match.visible @@ -686,7 +714,7 @@ def persist_hdas(elements, model_persistence_context, final_job_state='ok'): if "elements" in element: collect_elements_for_history(element["elements"]) else: - discovered_file = discovered_file_for_element(element, model_persistence_context.job_working_directory) + discovered_file = discovered_file_for_element(element, model_persistence_context) fields_match = discovered_file.match designation = fields_match.designation ext = fields_match.ext @@ -795,9 +823,10 @@ def replace_request_syntax_sugar(obj): DiscoveredFile = namedtuple('DiscoveredFile', ['path', 'collector', 'match']) -def discovered_file_for_element(dataset, job_working_directory, parent_identifiers=None, collector=None): +def discovered_file_for_element(dataset, model_persistence_context: Union['JobContext', 'SessionlessJobContext', SessionlessModelPersistenceContext], parent_identifiers=None, collector=None): + model_persistence_context.increment_discovered_file_count() parent_identifiers = parent_identifiers or [] - target_directory = discover_target_directory(getattr(collector, "directory", None), job_working_directory) + target_directory = discover_target_directory(getattr(collector, "directory", None), model_persistence_context.job_working_directory) filename = dataset.get("filename") error_message = dataset.get("error_message") if error_message is None: diff --git a/lib/galaxy/tools/__init__.py b/lib/galaxy/tools/__init__.py index af64520a9bc..af3405daae7 100644 --- a/lib/galaxy/tools/__init__.py +++ b/lib/galaxy/tools/__init__.py @@ -2114,6 +2114,7 @@ class Tool(Dictifiable): object_store=tool.app.object_store, final_job_state=final_job_state, flush_per_n_datasets=tool.app.config.flush_per_n_datasets, + max_discovered_files=tool.app.config.max_discovered_files, ) collected = output_collect.collect_primary_datasets( job_context, diff --git a/lib/galaxy/tools/actions/metadata.py b/lib/galaxy/tools/actions/metadata.py index 15fb56ac7be..2c7cb1e925c 100644 --- a/lib/galaxy/tools/actions/metadata.py +++ b/lib/galaxy/tools/actions/metadata.py @@ -103,6 +103,7 @@ class SetMetadataToolAction(ToolAction): job_metadata=os.path.join(job_working_dir, 'working', tool.provided_metadata_file), include_command=False, max_metadata_value_size=app.config.max_metadata_value_size, + max_discovered_files=app.config.max_discovered_files, validate_outputs=validate_outputs, job=job, kwds={'overwrite': overwrite}) diff --git a/lib/galaxy/webapps/galaxy/config_schema.yml b/lib/galaxy/webapps/galaxy/config_schema.yml index 6af6babf1e5..2c0ddefc257 100644 --- a/lib/galaxy/webapps/galaxy/config_schema.yml +++ b/lib/galaxy/webapps/galaxy/config_schema.yml @@ -2623,6 +2623,16 @@ mapping: Higher values will lead to fewer database flushes and faster execution, but require more memory. Set to -1 to disable creating datasets in batches. + max_discovered_files: + type: int + default: 10000 + required: false + desc: | + Set this to a positive integer value to limit the number of datasets that can be discovered by + a single job. This prevents accidentally creating large numbers of datasets when running tools + that create a potentially unlimited number of output datasets, such as tools that split a file + into a collection of datasets for each line in an input dataset. + history_local_serial_workflow_scheduling: type: bool default: false diff --git a/test/integration/test_max_discovered_files.py b/test/integration/test_max_discovered_files.py new file mode 100644 index 00000000000..979685213ee --- /dev/null +++ b/test/integration/test_max_discovered_files.py @@ -0,0 +1,37 @@ +"""Integration tests for max_discoverd_files setting.""" +from galaxy_test.base.populators import DatasetPopulator +from galaxy_test.driver import integration_util + + +class MaxDiscoveredFilesTestCase(integration_util.IntegrationTestCase): + """Describe a Galaxy test instance with embedded pulsar configured.""" + + framework_tool_and_types = True + max_discovered_files = 9 + + @classmethod + def handle_galaxy_config_kwds(cls, config): + config["max_discovered_files"] = cls.max_discovered_files + + def setUp(self): + super().setUp() + self.dataset_populator = DatasetPopulator(self.galaxy_interactor) + + def test_discover(self): + with self.dataset_populator.test_history() as history_id: + response = self.dataset_populator.run_tool("discover_sort_by", inputs={}, history_id=history_id) + job_id = response["jobs"][0]['id'] + self.dataset_populator.wait_for_job(job_id, assert_ok=False) + job_details_response = self.dataset_populator.get_job_details(job_id, full=True) + job_details_response.raise_for_status() + job_details = job_details_response.json() + assert job_details['state'] == 'error' + assert f"Job generated more than maximum number ({self.max_discovered_files}) of output datasets" in job_details['job_messages'] + + +class ExtendedMetadataMaxDiscoveredFilesTestCase(MaxDiscoveredFilesTestCase): + + @classmethod + def handle_galaxy_config_kwds(cls, config): + config["max_discovered_files"] = cls.max_discovered_files + config["metadata_strategy"] = "extended" diff --git a/test/unit/app/jobs/test_job_context.py b/test/unit/app/jobs/test_job_context.py index 019ab2e87df..7e0821bd081 100644 --- a/test/unit/app/jobs/test_job_context.py +++ b/test/unit/app/jobs/test_job_context.py @@ -66,7 +66,7 @@ def test_job_context_discover_outputs_flushes_once(mocker): collection_description = FilePatternDatasetCollectionDescription(pattern="__name__") collection = model.DatasetCollection(collection_type='list', populated=False) sa_session.add(collection) - job_context = JobContext(tool, tool_provided_metadata, job, job_working_directory, permission_provider, metadata_source_provider, input_dbkey, object_store, final_job_state) + job_context = JobContext(tool, tool_provided_metadata, job, job_working_directory, permission_provider, metadata_source_provider, input_dbkey, object_store, final_job_state, max_discovered_files=100) collection_builder = builder.BoundCollectionBuilder(collection) dataset_collectors = [dataset_collector(collection_description)] output_name = 'output'