Merge pull request #13170 from mvdbeek/max_discovered_files

Limit maximum number of discovered files per job

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``
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

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):

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

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

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)

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,
         }
 

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)

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()

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.

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:

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,

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})

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

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"

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'