From 90c01d551d2e91f90efb136ca80f6f11e7d57b5a Mon Sep 17 00:00:00 2001 From: Nate Coraor Date: Thu, 7 May 2015 12:21:11 -0400 Subject: [PATCH] Add an `is_valid` column to job_external_output_metadata so that the temp files can be regenerated when the job is resubmitted. --- lib/galaxy/datatypes/metadata.py | 18 ++++++- lib/galaxy/jobs/__init__.py | 7 +++ .../jobs/runners/state_handlers/resubmit.py | 1 + lib/galaxy/model/mapping.py | 1 + ...9_job_external_output_metadata_validity.py | 50 +++++++++++++++++++ 5 files changed, 75 insertions(+), 2 deletions(-) create mode 100644 lib/galaxy/model/migrate/versions/0129_job_external_output_metadata_validity.py diff --git a/lib/galaxy/datatypes/metadata.py b/lib/galaxy/datatypes/metadata.py index 0016cb30624..e30ac336c66 100644 --- a/lib/galaxy/datatypes/metadata.py +++ b/lib/galaxy/datatypes/metadata.py @@ -688,11 +688,15 @@ class JobExternalOutputMetadataWrapper( object ): def get_output_filenames_by_dataset( self, dataset, sa_session ): if isinstance( dataset, galaxy.model.HistoryDatasetAssociation ): return sa_session.query( galaxy.model.JobExternalOutputMetadata ) \ - .filter_by( job_id=self.job_id, history_dataset_association_id=dataset.id ) \ + .filter_by( job_id=self.job_id, + history_dataset_association_id=dataset.id, + is_valid=True ) \ .first() # there should only be one or None elif isinstance( dataset, galaxy.model.LibraryDatasetDatasetAssociation ): return sa_session.query( galaxy.model.JobExternalOutputMetadata ) \ - .filter_by( job_id=self.job_id, library_dataset_dataset_association_id=dataset.id ) \ + .filter_by( job_id=self.job_id, + library_dataset_dataset_association_id=dataset.id, + is_valid=True ) \ .first() # there should only be one or None return None @@ -701,6 +705,16 @@ class JobExternalOutputMetadataWrapper( object ): # need to make different keys for them, since ids can overlap return "%s_%d" % ( dataset.__class__.__name__, dataset.id ) + def invalidate_external_metadata( self, datasets, sa_session ): + for dataset in datasets: + jeom = self.get_output_filenames_by_dataset( dataset, sa_session ) + # shouldn't be more than one valid, but you never know + while jeom: + jeom.is_valid = False + sa_session.add( jeom ) + sa_session.flush() + jeom = self.get_output_filenames_by_dataset( dataset, sa_session ) + def setup_external_metadata( self, datasets, sa_session, exec_dir=None, tmp_dir=None, dataset_files_path=None, output_fnames=None, config_root=None, diff --git a/lib/galaxy/jobs/__init__.py b/lib/galaxy/jobs/__init__.py index ca8e84c3a3d..c2cfe8c616e 100644 --- a/lib/galaxy/jobs/__init__.py +++ b/lib/galaxy/jobs/__init__.py @@ -1555,6 +1555,13 @@ class JobWrapper( object ): return ExpressionContext( meta, job_context ) return job_context + def invalidate_external_metadata( self ): + job = self.get_job() + self.external_output_metadata.invalidate_external_metadata( [ output_dataset_assoc.dataset for + output_dataset_assoc in + job.output_datasets + job.output_library_datasets ], + self.sa_session ) + def setup_external_metadata( self, exec_dir=None, tmp_dir=None, dataset_files_path=None, config_root=None, config_file=None, datatypes_config=None, diff --git a/lib/galaxy/jobs/runners/state_handlers/resubmit.py b/lib/galaxy/jobs/runners/state_handlers/resubmit.py index f1d48b09942..c62d24a9287 100644 --- a/lib/galaxy/jobs/runners/state_handlers/resubmit.py +++ b/lib/galaxy/jobs/runners/state_handlers/resubmit.py @@ -43,6 +43,7 @@ def failure(app, job_runner, job_state): .cache_job_destination(new_destination)) # Reset job state job_state.job_wrapper.clear_working_directory() + job_state.job_wrapper.invalidate_external_metadata() job = job_state.job_wrapper.get_job() if resubmit.get('handler', None): log.debug('(%s/%s) Job reassigned to handler %s', diff --git a/lib/galaxy/model/mapping.py b/lib/galaxy/model/mapping.py index 55ab8c1019e..a5f3b9b1db8 100644 --- a/lib/galaxy/model/mapping.py +++ b/lib/galaxy/model/mapping.py @@ -485,6 +485,7 @@ model.JobExternalOutputMetadata.table = Table( "job_external_output_metadata", m Column( "job_id", Integer, ForeignKey( "job.id" ), index=True ), Column( "history_dataset_association_id", Integer, ForeignKey( "history_dataset_association.id" ), index=True, nullable=True ), Column( "library_dataset_dataset_association_id", Integer, ForeignKey( "library_dataset_dataset_association.id" ), index=True, nullable=True ), + Column( "is_valid", Boolean, default=True ), Column( "filename_in", String( 255 ) ), Column( "filename_out", String( 255 ) ), Column( "filename_results_code", String( 255 ) ), diff --git a/lib/galaxy/model/migrate/versions/0129_job_external_output_metadata_validity.py b/lib/galaxy/model/migrate/versions/0129_job_external_output_metadata_validity.py new file mode 100644 index 00000000000..f24a99f61b2 --- /dev/null +++ b/lib/galaxy/model/migrate/versions/0129_job_external_output_metadata_validity.py @@ -0,0 +1,50 @@ +""" +Migration script to allow invalidation of job external output metadata temp files +""" +from sqlalchemy import * +from sqlalchemy.orm import * +from migrate import * +from migrate.changeset import * +from galaxy.model.custom_types import * + +import datetime +now = datetime.datetime.utcnow + +import logging +log = logging.getLogger( __name__ ) + +metadata = MetaData() + + +def upgrade(migrate_engine): + metadata.bind = migrate_engine + print __doc__ + metadata.reflect() + + isvalid_column = Column( "is_valid", Boolean, default=True ) + __add_column( isvalid_column, "job_external_output_metadata", metadata ) + + +def downgrade(migrate_engine): + metadata.bind = migrate_engine + metadata.reflect() + + __drop_column( isvalid_column, "job_external_output_metadata", metadata ) + + +def __add_column(column, table_name, metadata, **kwds): + try: + table = Table( table_name, metadata, autoload=True ) + column.create( table, **kwds ) + except Exception as e: + print str(e) + log.exception( "Adding column %s failed." % column) + + +def __drop_column( column_name, table_name, metadata ): + try: + table = Table( table_name, metadata, autoload=True ) + getattr( table.c, column_name ).drop() + except Exception as e: + print str(e) + log.exception( "Dropping column %s failed." % column_name )