diff --git a/lib/galaxy/jobs/actions/post.py b/lib/galaxy/jobs/actions/post.py index 267a765e66f..138d4b8005b 100644 --- a/lib/galaxy/jobs/actions/post.py +++ b/lib/galaxy/jobs/actions/post.py @@ -269,10 +269,10 @@ class DeleteIntermediatesAction(DefaultJobAction): # concurrently, sometimes non-terminal steps won't be cleaned up # because of the lag in job state updates. sa_session.flush() - if not job.workflow_invocation_step: + if not job.workflow_invocation_step_assoc.workflow_invocation_step: log.debug("This job is not part of a workflow invocation, delete intermediates aborted.") return - wfi = job.workflow_invocation_step.workflow_invocation + wfi = job.workflow_invocation_step_assoc.workflow_invocation_step.workflow_invocation sa_session.refresh(wfi) if wfi.active: log.debug("Workflow still scheduling so new jobs may appear, skipping deletion of intermediate files.") @@ -285,9 +285,9 @@ class DeleteIntermediatesAction(DefaultJobAction): jobs_to_check = [] for wfi_step in wfi_steps: sa_session.refresh(wfi_step) - wfi_step_job = wfi_step.job - if wfi_step_job: - jobs_to_check.append(wfi_step_job) + wfi_step_job_assocs = wfi_step.jobs + if wfi_step_job_assocs: + jobs_to_check.extend(map(lambda j: j.job, wfi_step_job_assocs)) else: log.debug("No job found yet for wfi_step %s, (step %s)" % (wfi_step, wfi_step.workflow_step)) for j2c in jobs_to_check: @@ -302,7 +302,7 @@ class DeleteIntermediatesAction(DefaultJobAction): for (input_dataset, creating_job) in creating_jobs: sa_session.refresh(creating_job) sa_session.refresh(input_dataset) - for input_dataset in [x.dataset for (x, creating_job) in creating_jobs if creating_job.workflow_invocation_step and creating_job.workflow_invocation_step.workflow_invocation == wfi]: + for input_dataset in [x.dataset for (x, creating_job) in creating_jobs if creating_job.workflow_invocation_step_assoc and creating_job.workflow_invocation_step_assoc.workflow_invocation_step.workflow_invocation == wfi]: # note that the above input_dataset is a reference to a # job.input_dataset.dataset at this point safe_to_delete = True diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py index f054edac4e3..707f96cbdbf 100644 --- a/lib/galaxy/model/__init__.py +++ b/lib/galaxy/model/__init__.py @@ -803,8 +803,9 @@ class Job(object, JobLike, UsesCreateAndUpdateTime, Dictifiable): def set_final_state(self, final_state): self.set_state(final_state) - if self.workflow_invocation_step: - self.workflow_invocation_step.update() + workflow_invocation_step_assoc = self.workflow_invocation_step_assoc + if workflow_invocation_step_assoc: + workflow_invocation_step_assoc.workflow_invocation_step.update() def get_destination_configuration(self, config, key, default=None): """ Get a destination parameter that can be defaulted back @@ -4036,17 +4037,16 @@ class WorkflowInvocation(object, UsesCreateAndUpdateTime, Dictifiable): step_invocations = {} for invocation_step in self.steps: step_id = invocation_step.workflow_step_id - if step_id not in step_invocations: - step_invocations[step_id] = [] - step_invocations[step_id].append(invocation_step) + assert step_id not in step_invocations + step_invocations[step_id] = invocation_step return step_invocations - def step_invocations_for_step_id(self, step_id): - step_invocations = [] + def step_invocation_for_step_id(self, step_id): + target_invocation_step = None for invocation_step in self.steps: if step_id == invocation_step.workflow_step_id: - step_invocations.append(invocation_step) - return step_invocations + target_invocation_step = invocation_step + return target_invocation_step @staticmethod def poll_active_workflow_ids( @@ -4072,6 +4072,24 @@ class WorkflowInvocation(object, UsesCreateAndUpdateTime, Dictifiable): # is relatively intutitive. return [wid for wid in query.all()] + def add_output(self, workflow_output, step, output_object): + if output_object.history_content_type == "dataset": + output_assoc = WorkflowInvocationOutputDatasetAssociation() + output_assoc.workflow_invocation = self + output_assoc.workflow_output = workflow_output + output_assoc.workflow_step = step + output_assoc.dataset = output_object + self.output_datasets.append(output_assoc) + elif output_object.history_content_type == "dataset_collection": + output_assoc = WorkflowInvocationOutputDatasetCollectionAssociation() + output_assoc.workflow_invocation = self + output_assoc.workflow_output = workflow_output + output_assoc.workflow_step = step + output_assoc.dataset_collection = output_object + self.output_dataset_collections.append(output_assoc) + else: + raise Exception("Uknown output type encountered") + def to_dict(self, view='collection', value_mapper=None, step_details=False): rval = super(WorkflowInvocation, self).to_dict(view=view, value_mapper=value_mapper) if view == 'element': @@ -4087,17 +4105,43 @@ class WorkflowInvocation(object, UsesCreateAndUpdateTime, Dictifiable): inputs = {} for step in self.steps: if step.workflow_step.type == 'tool': - for step_input in step.workflow_step.input_connections: - output_step_type = step_input.output_step.type - if output_step_type in ['data_input', 'data_collection_input']: - src = "hda" if output_step_type == 'data_input' else 'hdca' - for job_input in step.job.input_datasets: - if job_input.name == step_input.input_name: - inputs[str(step_input.output_step.order_index)] = { - "id": job_input.dataset_id, "src": src, - "uuid" : str(job_input.dataset.dataset.uuid) if job_input.dataset.dataset.uuid is not None else None - } + for step_job_assoc in step.jobs: + for step_input in step.workflow_step.input_connections: + output_step_type = step_input.output_step.type + if output_step_type in ['data_input', 'data_collection_input']: + src = "hda" if output_step_type == 'data_input' else 'hdca' + for job_input in step_job_assoc.job.input_datasets: + if job_input.name == step_input.input_name: + inputs[str(step_input.output_step.order_index)] = { + "id": job_input.dataset_id, "src": src, + "uuid" : str(job_input.dataset.dataset.uuid) if job_input.dataset.dataset.uuid is not None else None + } rval['inputs'] = inputs + + outputs = {} + for output_assoc in self.output_datasets: + label = output_assoc.workflow_output.label + if not label: + continue + + outputs[label] = { + 'src': 'hda', + 'id': output_assoc.dataset_id, + } + + output_collections = {} + for output_assoc in self.output_dataset_collections: + label = output_assoc.workflow_output.label + if not label: + continue + + output_collections[label] = { + 'src': 'hdca', + 'id': output_assoc.dataset_collection_id, + } + + rval['outputs'] = outputs + rval['output_collections'] = output_collections return rval def update(self): @@ -4137,36 +4181,70 @@ class WorkflowInvocationToSubworkflowInvocationAssociation(object, Dictifiable): class WorkflowInvocationStep(object, Dictifiable): - dict_collection_visible_keys = ['id', 'update_time', 'job_id', 'workflow_step_id', 'action'] - dict_element_visible_keys = ['id', 'update_time', 'job_id', 'workflow_step_id', 'action'] + dict_collection_visible_keys = ['id', 'update_time', 'job_id', 'workflow_step_id', 'state', 'action'] + dict_element_visible_keys = ['id', 'update_time', 'job_id', 'workflow_step_id', 'state', 'action'] + states = Bunch( + NEW='new', # Brand new workflow invocation step + READY='ready', # Workflow invocation step ready for another iteration of scheduling. + SCHEDULED='scheduled', # Workflow invocation step has been scheduled. + # CANCELLED='cancelled', TODO: implement and expose + # FAILED='failed', TODO: implement and expose + ) def update(self): self.workflow_invocation.update() + def add_output(self, output_name, output_object): + if output_object.history_content_type == "dataset": + output_assoc = WorkflowInvocationStepOutputDatasetAssociation() + output_assoc.workflow_invocation_step = self + output_assoc.dataset = output_object + output_assoc.output_name = output_name + self.output_datasets.append(output_assoc) + elif output_object.history_content_type == "dataset_collection": + output_assoc = WorkflowInvocationStepOutputDatasetCollectionAssociation() + output_assoc.workflow_invocation_step = self + output_assoc.dataset_collection = output_object + output_assoc.output_name = output_name + self.output_dataset_collections.append(output_assoc) + else: + raise Exception("Uknown output type encountered") + def to_dict(self, view='collection', value_mapper=None): rval = super(WorkflowInvocationStep, self).to_dict(view=view, value_mapper=value_mapper) rval['order_index'] = self.workflow_step.order_index rval['workflow_step_label'] = self.workflow_step.label rval['workflow_step_uuid'] = str(self.workflow_step.uuid) - rval['state'] = self.job.state if self.job is not None else None - if self.job is not None and view == 'element': - output_dict = {} - for i in self.job.output_datasets: - if i.dataset is not None: - output_dict[i.name] = { - "id" : i.dataset.id, "src" : "hda", - "uuid" : str(i.dataset.dataset.uuid) if i.dataset.dataset.uuid is not None else None - } - for i in self.job.output_library_datasets: - if i.dataset is not None: - output_dict[i.name] = { - "id" : i.dataset.id, "src" : "ldda", - "uuid" : str(i.dataset.dataset.uuid) if i.dataset.dataset.uuid is not None else None - } - rval['outputs'] = output_dict + # Following no longer makes sense... + # rval['state'] = self.job.state if self.job is not None else None + if view == 'element': + outputs = {} + for output_assoc in self.output_datasets: + name = output_assoc.output_name + outputs[name] = { + 'src': 'hda', + 'id': output_assoc.dataset.id, + 'uuid': str(output_assoc.dataset.dataset.uuid) if output_assoc.dataset.dataset.uuid is not None else None + } + + output_collections = {} + for output_assoc in self.output_dataset_collections: + name = output_assoc.output_name + output_collections[name] = { + 'src': 'hdca', + 'id': output_assoc.dataset_collection.id, + } + + rval['outputs'] = outputs + rval['output_collections'] = output_collections return rval +class WorkflowInvocationStepJobAssociation(object, Dictifiable): + dict_collection_visible_keys = ('id', 'job_id', 'workflow_invocation_step_id') + dict_element_visible_keys = ('id', 'job_id', 'workflow_invocation_step_id') + + class WorkflowRequest(object, Dictifiable): dict_collection_visible_keys = ['id', 'name', 'type', 'state', 'history_id', 'workflow_id'] dict_element_visible_keys = ['id', 'name', 'type', 'state', 'history_id', 'workflow_id'] @@ -4221,6 +4299,26 @@ class WorkflowRequestInputStepParmeter(object, Dictifiable): dict_collection_visible_keys = ['id', 'workflow_invocation_id', 'workflow_step_id', 'parameter_value'] +class WorkflowInvocationOutputDatasetAssociation(object, Dictifiable): + """Represents links to output datasets for the workflow.""" + dict_collection_visible_keys = ['id', 'workflow_invocation_id', 'workflow_step_id', 'dataset_id', 'name'] + + +class WorkflowInvocationOutputDatasetCollectionAssociation(object, Dictifiable): + """Represents links to output dataset collections for the workflow.""" + dict_collection_visible_keys = ['id', 'workflow_invocation_id', 'workflow_step_id', 'dataset_collection_id', 'name'] + + +class WorkflowInvocationStepOutputDatasetAssociation(object, Dictifiable): + """Represents links to output datasets for the workflow.""" + dict_collection_visible_keys = ['id', 'workflow_invocation_step_id', 'dataset_id', 'output_name'] + + +class WorkflowInvocationStepOutputDatasetCollectionAssociation(object, Dictifiable): + """Represents links to output dataset collections for the workflow.""" + dict_collection_visible_keys = ['id', 'workflow_invocation_step_id', 'dataset_collection_id', 'output_name'] + + class MetadataFile(StorableObject): def __init__(self, dataset=None, name=None): diff --git a/lib/galaxy/model/mapping.py b/lib/galaxy/model/mapping.py index 97923d892cc..49d6d4ed206 100644 --- a/lib/galaxy/model/mapping.py +++ b/lib/galaxy/model/mapping.py @@ -901,9 +901,54 @@ model.WorkflowInvocationStep.table = Table( Column("update_time", DateTime, default=now, onupdate=now), Column("workflow_invocation_id", Integer, ForeignKey("workflow_invocation.id"), index=True, nullable=False), Column("workflow_step_id", Integer, ForeignKey("workflow_step.id"), index=True, nullable=False), - Column("job_id", Integer, ForeignKey("job.id"), index=True, nullable=True), + Column("state", TrimmedString(64), index=True), Column("action", JSONType, nullable=True)) + +model.WorkflowInvocationStepJobAssociation.table = Table( + "workflow_invocation_step_job_association", metadata, + Column("id", Integer, primary_key=True), + Column("workflow_invocation_step_id", Integer, ForeignKey("workflow_invocation_step.id"), index=True, nullable=False), + Column("order_index", Integer, nullable=True), # recovering partially complete WorkflowInvocationSteps requires knowing which jobs have been scheduled + Column("job_id", Integer, ForeignKey("job.id"), index=True, nullable=False), +) + + +model.WorkflowInvocationOutputDatasetAssociation.table = Table( + "workflow_invocation_output_dataset_association", metadata, + Column("id", Integer, primary_key=True), + Column("workflow_invocation_id", Integer, ForeignKey("workflow_invocation.id"), index=True), + Column("workflow_step_id", Integer, ForeignKey("workflow_step.id"), index=True), + Column("dataset_id", Integer, ForeignKey("history_dataset_association.id"), index=True), + Column("workflow_output_id", Integer, ForeignKey("workflow_output.id"), index=True), +) + +model.WorkflowInvocationOutputDatasetCollectionAssociation.table = Table( + "workflow_invocation_output_dataset_collection_association", metadata, + Column("id", Integer, primary_key=True), + Column("workflow_invocation_id", Integer, ForeignKey("workflow_invocation.id"), index=True), + Column("workflow_step_id", Integer, ForeignKey("workflow_step.id"), index=True), + Column("dataset_collection_id", Integer, ForeignKey("history_dataset_collection_association.id"), index=True), + Column("workflow_output_id", Integer, ForeignKey("workflow_output.id"), index=True), +) + +model.WorkflowInvocationStepOutputDatasetAssociation.table = Table( + "workflow_invocation_step_output_dataset_association", metadata, + Column("id", Integer, primary_key=True), + Column("workflow_invocation_step_id", Integer, ForeignKey("workflow_invocation_step.id"), index=True), + Column("dataset_id", Integer, ForeignKey("history_dataset_association.id"), index=True), + Column("output_name", String(255), nullable=True), +) + +model.WorkflowInvocationStepOutputDatasetCollectionAssociation.table = Table( + "workflow_invocation_step_output_dataset_collection_association", metadata, + Column("id", Integer, primary_key=True), + Column("workflow_invocation_step_id", Integer, ForeignKey("workflow_invocation_step.id"), index=True), + Column("workflow_step_id", Integer, ForeignKey("workflow_step.id"), index=True), + Column("dataset_collection_id", Integer, ForeignKey("history_dataset_collection_association.id"), index=True), + Column("output_name", String(255), nullable=True), +) + model.WorkflowInvocationToSubworkflowInvocationAssociation.table = Table( "workflow_invocation_to_subworkflow_invocation_association", metadata, Column("id", Integer, primary_key=True), @@ -2310,7 +2355,7 @@ mapper(model.WorkflowInvocation, model.WorkflowInvocation.table, properties=dict uselist=True, ), steps=relation(model.WorkflowInvocationStep, - backref='workflow_invocation'), + backref="workflow_invocation"), workflow=relation(model.Workflow) )) @@ -2323,12 +2368,17 @@ mapper(model.WorkflowInvocationToSubworkflowInvocationAssociation, model.Workflo workflow_step=relation(model.WorkflowStep), )) -mapper(model.WorkflowInvocationStep, model.WorkflowInvocationStep.table, properties=dict( - workflow_step=relation(model.WorkflowStep), + +simple_mapping(model.WorkflowInvocationStepJobAssociation, + workflow_invocation_step=relation(model.WorkflowInvocationStep, backref="jobs"), job=relation(model.Job, - backref=backref('workflow_invocation_step', - uselist=False)) -)) + backref=backref('workflow_invocation_step_assoc', + uselist=False))) + + +simple_mapping(model.WorkflowInvocationStep, + workflow_step=relation(model.WorkflowStep)) + simple_mapping(model.WorkflowRequestInputParameter, workflow_invocation=relation(model.WorkflowInvocation)) @@ -2358,6 +2408,39 @@ mapper(model.MetadataFile, model.MetadataFile.table, properties=dict( library_dataset=relation(model.LibraryDatasetDatasetAssociation) )) + +simple_mapping( + model.WorkflowInvocationOutputDatasetAssociation, + workflow_invocation=relation(model.WorkflowInvocation, backref="output_datasets"), + workflow_step=relation(model.WorkflowStep), + dataset=relation(model.HistoryDatasetAssociation), + workflow_output=relation(model.WorkflowOutput), +) + + +simple_mapping( + model.WorkflowInvocationOutputDatasetCollectionAssociation, + workflow_invocation=relation(model.WorkflowInvocation, backref="output_dataset_collections"), + workflow_step=relation(model.WorkflowStep), + dataset_collection=relation(model.HistoryDatasetCollectionAssociation), + workflow_output=relation(model.WorkflowOutput), +) + + +simple_mapping( + model.WorkflowInvocationStepOutputDatasetAssociation, + workflow_invocation_step=relation(model.WorkflowInvocationStep, backref="output_datasets"), + dataset=relation(model.HistoryDatasetAssociation), +) + + +simple_mapping( + model.WorkflowInvocationStepOutputDatasetCollectionAssociation, + workflow_invocation_step=relation(model.WorkflowInvocationStep, backref="output_dataset_collections"), + dataset_collection=relation(model.HistoryDatasetCollectionAssociation), +) + + mapper(model.PageRevision, model.PageRevision.table) mapper(model.Page, model.Page.table, properties=dict( diff --git a/lib/galaxy/model/migrate/versions/0136_record_workflow_outputs.py b/lib/galaxy/model/migrate/versions/0136_record_workflow_outputs.py new file mode 100644 index 00000000000..1b83a49aa31 --- /dev/null +++ b/lib/galaxy/model/migrate/versions/0136_record_workflow_outputs.py @@ -0,0 +1,185 @@ +""" +Migration script for workflow request tables. +""" +from __future__ import print_function + +import datetime +import logging + +from collections import OrderedDict + +from migrate.changeset.constraint import ForeignKeyConstraint +from sqlalchemy import Column, DateTime, ForeignKey, Integer, MetaData, String, Table + +from galaxy.model.custom_types import JSONType, TrimmedString + + +now = datetime.datetime.utcnow + +log = logging.getLogger(__name__) +metadata = MetaData() + + +def get_new_tables(): + # Normally we define this globally in the file, but we need to delay the + # reading of existing tables because an existing workflow_invocation_step + # table exists that we want to recreate. + + workflow_invocation_output_dataset_association_table = Table( + "workflow_invocation_output_dataset_association", metadata, + Column("id", Integer, primary_key=True), + Column("workflow_invocation_id", Integer, ForeignKey("workflow_invocation.id"), index=True), + Column("workflow_step_id", Integer, ForeignKey("workflow_step.id")), + Column("dataset_id", Integer, ForeignKey("history_dataset_association.id"), index=True), + Column("workflow_output_id", Integer, ForeignKey("workflow_output.id")), + ) + + workflow_invocation_output_dataset_collection_association_table = Table( + "workflow_invocation_output_dataset_collection_association", metadata, + Column("id", Integer, primary_key=True), + Column("workflow_invocation_id", Integer, ForeignKey("workflow_invocation.id"), index=True), + Column("workflow_step_id", Integer, ForeignKey("workflow_step.id")), + Column("dataset_collection_id", Integer, ForeignKey("history_dataset_collection_association.id"), index=True), + Column("workflow_output_id", Integer, ForeignKey("workflow_output.id")), + ) + + workflow_invocation_step_output_dataset_association_table = Table( + "workflow_invocation_step_output_dataset_association", metadata, + Column("id", Integer, primary_key=True), + Column("workflow_invocation_step_id", Integer, ForeignKey("workflow_invocation_step.id"), index=True), + Column("dataset_id", Integer, ForeignKey("history_dataset_association.id"), index=True), + Column("output_name", String(255), nullable=True), + ) + + workflow_invocation_step_output_dataset_collection_association_table = Table( + "workflow_invocation_step_output_dataset_collection_association", metadata, + Column("id", Integer, primary_key=True), + Column("workflow_invocation_step_id", Integer, ForeignKey("workflow_invocation_step.id"), index=True), + Column("workflow_step_id", Integer, ForeignKey("workflow_step.id")), + Column("dataset_collection_id", Integer, ForeignKey("history_dataset_collection_association.id"), index=True), + Column("output_name", String(255), nullable=True), + ) + + workflow_invocation_step_table = Table( + "workflow_invocation_step", metadata, + Column("id", Integer, primary_key=True), + Column("create_time", DateTime, default=now), + Column("update_time", DateTime, default=now, onupdate=now), + Column("workflow_invocation_id", Integer, ForeignKey("workflow_invocation.id"), index=True, nullable=False), + Column("workflow_step_id", Integer, ForeignKey("workflow_step.id"), index=True, nullable=False), + Column("action", JSONType, nullable=True), + Column("state", TrimmedString(64), default="new"), + ) + + workflow_invocation_step_job_association_table = Table( + "workflow_invocation_step_job_association", metadata, + Column("id", Integer, primary_key=True), + Column("workflow_invocation_step_id", Integer, ForeignKey("workflow_invocation_step.id"), index=True, nullable=False), + Column("order_index", Integer, nullable=True), + Column("job_id", Integer, ForeignKey("job.id"), index=True, nullable=False), + ) + + tables = OrderedDict() + tables["workflow_invocation_step"] = workflow_invocation_step_table + tables["workflow_invocation_output_dataset_association"] = workflow_invocation_output_dataset_association_table + tables["workflow_invocation_output_dataset_collection_association"] = workflow_invocation_output_dataset_collection_association_table + tables["workflow_invocation_step_output_dataset_association"] = workflow_invocation_step_output_dataset_association_table + tables["workflow_invocation_step_output_dataset_collection_association"] = workflow_invocation_step_output_dataset_collection_association_table + tables["workflow_invocation_step_job_association"] = workflow_invocation_step_job_association_table + + return tables + + +def upgrade(migrate_engine): + metadata.bind = migrate_engine + print(__doc__) + + LegacyWorkflowInvocationStep_table = Table("workflow_invocation_step", metadata, autoload=True) + ExistingWorkflowInvocation_table = Table("workflow_invocation", metadata, autoload=True) + + cons = ForeignKeyConstraint([LegacyWorkflowInvocationStep_table.c.workflow_invocation_id], [ExistingWorkflowInvocation_table.c.id]) + cons.drop() + + for index in LegacyWorkflowInvocationStep_table.indexes: + index.drop() + + LegacyWorkflowInvocationStep_table.rename("workflow_invocation_step_premigrate135") + # Try to deregister that workflow_invocation_step to work around some caching problems + # it seems. + LegacyWorkflowInvocationStep_table.deregister() + metadata._remove_table("workflow_invocation_step", metadata.schema) + + metadata.reflect() + tables = get_new_tables() + for table in tables.values(): + __create(table) + + def nextval(table, col='id'): + if migrate_engine.name in ['postgres', 'postgresql']: + return "nextval('%s_%s_seq')" % (table, col) + elif migrate_engine.name in ['mysql', 'sqlite']: + return "null" + else: + raise Exception("Unhandled database type") + + # Skips action - since I can't aggregate (sql needs a RANDOM(col)) that and it is only used by optional + # beta extensions. + cmd = \ + "INSERT INTO workflow_invocation_step " + \ + "(id, create_time, update_time, workflow_invocation_id, workflow_step_id, action, state)" + \ + "SELECT " + \ + nextval('workflow_invocation_step') + " AS id, " \ + "MIN(create_time) AS create_time, " + \ + "MAX(update_time) AS update_time, " + \ + "workflow_invocation_step_premigrate135.workflow_invocation_id AS workflow_invocation_id, " +\ + "workflow_invocation_step_premigrate135.workflow_step_id AS workflow_step_id, " + \ + "NULL AS action, " + \ + "'scheduled' AS state " + \ + "FROM workflow_invocation_step_premigrate135 " + \ + "WHERE workflow_invocation_step_premigrate135.workflow_step_id IS NOT NULL AND workflow_invocation_step_premigrate135.workflow_invocation_id IS NOT NULL " +\ + "GROUP BY workflow_invocation_step_premigrate135.workflow_invocation_id, workflow_invocation_step_premigrate135.workflow_step_id " + \ + "" + migrate_engine.execute(cmd) + + cmd = \ + "INSERT INTO workflow_invocation_step_job_association " + \ + "(id, workflow_invocation_step_id, order_index, job_id) " + \ + "SELECT " + \ + nextval('workflow_invocation_step_job_association') + " AS id, " \ + "workflow_invocation_step.id AS workflow_invocation_step_id, " + \ + "NULL AS order_index, " + \ + "job_id AS job_id " + \ + "FROM workflow_invocation_step_premigrate135 " + \ + "LEFT JOIN workflow_invocation_step on (" + \ + " workflow_invocation_step.workflow_invocation_id = workflow_invocation_step_premigrate135.workflow_invocation_id " + \ + " AND workflow_invocation_step.workflow_step_id = workflow_invocation_step_premigrate135.workflow_step_id" + \ + ") " + \ + "WHERE job_id is not NULL " + migrate_engine.execute(cmd) + + +def downgrade(migrate_engine): + metadata.bind = migrate_engine + metadata.reflect() + + tables = get_new_tables() + for table in tables.values(): + __drop(table) + + # Drop new workflow invocation step and job association table and restore legacy data. + LegacyWorkflowInvocationStep_table = Table("workflow_invocation_step_premigrate135", metadata, autoload=True) + LegacyWorkflowInvocationStep_table.rename("workflow_invocation_step") + + +def __create(table): + try: + table.create() + except Exception: + log.exception("Creating %s table failed.", table.name) + + +def __drop(table): + try: + table.drop() + except Exception: + log.exception("Dropping %s table failed.", table.name) diff --git a/lib/galaxy/tools/execute.py b/lib/galaxy/tools/execute.py index c24a57af1c0..e6d37538110 100644 --- a/lib/galaxy/tools/execute.py +++ b/lib/galaxy/tools/execute.py @@ -18,7 +18,13 @@ log = logging.getLogger(__name__) EXECUTION_SUCCESS_MESSAGE = "Tool [%s] created job [%s] %s" -def execute(trans, tool, param_combinations, history, rerun_remap_job_id=None, collection_info=None, workflow_invocation_uuid=None): +class PartialJobExecution(Exception): + + def __init__(self, jobs): + self.jobs = jobs + + +def execute(trans, tool, param_combinations, history, rerun_remap_job_id=None, collection_info=None, workflow_invocation_uuid=None, invocation_step=None, max_num_jobs=None): """ Execute a tool and return object containing summary (output data, number of failures, etc...). @@ -28,6 +34,8 @@ def execute(trans, tool, param_combinations, history, rerun_remap_job_id=None, c app = trans.app execution_cache = ToolExecutionCache(trans) + new_jobs = [] + def execute_single_job(params): job_timer = ExecutionTimer() if workflow_invocation_uuid: @@ -41,6 +49,7 @@ def execute(trans, tool, param_combinations, history, rerun_remap_job_id=None, c message = EXECUTION_SUCCESS_MESSAGE % (tool.id, job.id, job_timer) log.debug(message) execution_tracker.record_success(job, result) + new_jobs.append(job) else: execution_tracker.record_error(result) @@ -59,11 +68,27 @@ def execute(trans, tool, param_combinations, history, rerun_remap_job_id=None, c history ) + if invocation_step: + execution_tracker.recover_successful_jobs(invocation_step) + + previously_executed_jobs_count = len(execution_tracker.successful_jobs) job_count = len(execution_tracker.param_combinations) - if job_count < burst_at or burst_threads < 2: - for params in execution_tracker.param_combinations: - execute_single_job(params) + + jobs_executed = 0 + has_remaining_jobs = False + + if (job_count < burst_at or burst_threads < 2): + for index, params in enumerate(execution_tracker.param_combinations): + if index < previously_executed_jobs_count: + continue + elif max_num_jobs and jobs_executed >= max_num_jobs: + has_remaining_jobs = True + break + else: + execute_single_job(params) + jobs_executed += 1 else: + # TODO: re-record success... q = Queue() def worker(): @@ -77,11 +102,21 @@ def execute(trans, tool, param_combinations, history, rerun_remap_job_id=None, c t.daemon = True t.start() - for params in execution_tracker.param_combinations: - q.put(params) + for index, params in enumerate(execution_tracker.param_combinations): + if index < previously_executed_jobs_count: + continue + elif max_num_jobs and jobs_executed >= max_num_jobs: + has_remaining_jobs = True + break + else: + q.put(params) + jobs_executed += 1 q.join() + if has_remaining_jobs: + raise PartialJobExecution(new_jobs) + log.debug("Executed %d job(s) for tool %s request: %s" % (job_count, tool.id, all_jobs_timer)) if collection_info: history = history or tool.get_default_history_by_trans(trans) @@ -110,6 +145,17 @@ class ToolExecutionTracker(object): self.outputs_by_output_name = collections.defaultdict(list) self.implicit_collections = {} + def recover_successful_jobs(self, invocation_step): + # TODO: Optimize away the need to do this - we should just be dealing with IDs + # and such and we shouldn't fetch them until the very end when we need them to create + # collections. + for job_assoc in invocation_step.jobs: + job = job_assoc.job + for job_output in job.output_datasets: + self.outputs_by_output_name[job_output.name].append(job_output.dataset) + for job_output in job.output_dataset_collections: + self.outputs_by_output_name[job_output.name].append(job_output.dataset_collection) + def record_success(self, job, outputs): self.successful_jobs.append(job) self.output_datasets.extend(outputs) diff --git a/lib/galaxy/workflow/modules.py b/lib/galaxy/workflow/modules.py index d5fcbf641cf..7abc55636a5 100644 --- a/lib/galaxy/workflow/modules.py +++ b/lib/galaxy/workflow/modules.py @@ -21,7 +21,7 @@ from galaxy.tools import ( DefaultToolState, ToolInputsNotReadyException ) -from galaxy.tools.execute import execute +from galaxy.tools.execute import execute, PartialJobExecution from galaxy.tools.parameters import ( check_param, params_to_incoming, @@ -214,10 +214,14 @@ class WorkflowModule(object): state.decode(runtime_state, Bunch(inputs=self.get_runtime_inputs()), self.trans.app) return state - def execute(self, trans, progress, invocation, step): - """ Execute the given workflow step in the given workflow invocation. + def execute(self, trans, progress, invocation_step): + """ Execute the given workflow invocation step. + Use the supplied workflow progress object to track outputs, find - inputs, etc... + inputs, etc.... + + Return jobs created and a boolean indicating if there are additional + jobs to create on subsequent workflow scheduling tests. """ raise TypeError("Abstract method") @@ -230,11 +234,19 @@ class WorkflowModule(object): """ raise exceptions.RequestParameterInvalidException("Attempting to perform invocation step action on module that does not support actions.") - def recover_mapping(self, step, step_invocations, progress): + def recover_mapping(self, invocation_step, progress): """ Re-populate progress object with information about connections - from previously executed steps recorded via step_invocations. + from previously executed steps recorded via invocation_steps. """ - raise TypeError("Abstract method") + outputs = {} + + for output_dataset_assoc in invocation_step.output_datasets: + outputs[output_dataset_assoc.output_name] = output_dataset_assoc.dataset + + for output_dataset_collection_assoc in invocation_step.output_dataset_collections: + outputs[output_dataset_collection_assoc.output_name] = output_dataset_collection_assoc.dataset_collection + + progress.set_step_outputs(invocation_step, outputs, already_persisted=True) class SubWorkflowModule(WorkflowModule): @@ -320,11 +332,12 @@ class SubWorkflowModule(WorkflowModule): def get_content_id(self): return self.trans.security.encode_id(self.subworkflow.id) - def execute(self, trans, progress, invocation, step): + def execute(self, trans, progress, invocation_step): """ Execute the given workflow step in the given workflow invocation. Use the supplied workflow progress object to track outputs, find inputs, etc... """ + step = invocation_step.workflow_step subworkflow_invoker = progress.subworkflow_invoker(trans, step) subworkflow_invoker.invoke() subworkflow = subworkflow_invoker.workflow @@ -334,7 +347,7 @@ class SubWorkflowModule(WorkflowModule): workflow_output_label = workflow_output.label or "%s:%s" % (step.order_index, workflow_output.output_name) replacement = subworkflow_progress.get_replacement_workflow_output(workflow_output) outputs[workflow_output_label] = replacement - progress.set_step_outputs(step, outputs) + progress.set_step_outputs(invocation_step, outputs) return None def get_runtime_state(self): @@ -353,8 +366,10 @@ class InputModule(WorkflowModule): def get_data_inputs(self): return [] - def execute(self, trans, progress, invocation, step): - job, step_outputs = None, dict(output=step.state.inputs['input']) + def execute(self, trans, progress, invocation_step): + invocation = invocation_step.workflow_invocation + step = invocation_step.workflow_step + step_outputs = dict(output=step.state.inputs['input']) # Web controller may set copy_inputs_to_history, API controller always sets # inputs. @@ -378,11 +393,10 @@ class InputModule(WorkflowModule): content = next(iter(step_outputs.values())) if content: invocation.add_input(content, step.id) - progress.set_outputs_for_input(step, step_outputs) - return job + progress.set_outputs_for_input(invocation_step, step_outputs) - def recover_mapping(self, step, step_invocations, progress): - progress.set_outputs_for_input(step) + def recover_mapping(self, invocation_step, progress): + progress.set_outputs_for_input(invocation_step) class InputDataModule(InputModule): @@ -489,10 +503,10 @@ class InputParameterModule(WorkflowModule): def get_data_inputs(self): return [] - def execute(self, trans, progress, invocation, step): - job, step_outputs = None, dict(output=step.state.inputs['input']) - progress.set_outputs_for_input(step, step_outputs) - return job + def execute(self, trans, progress, invocation_step): + step = invocation_step.workflow_step + step_outputs = dict(output=step.state.inputs['input']) + progress.set_outputs_for_input(invocation_step, step_outputs) class PauseModule(WorkflowModule): @@ -520,18 +534,18 @@ class PauseModule(WorkflowModule): state.inputs = dict() return state - def execute(self, trans, progress, invocation, step): + def execute(self, trans, progress, invocation_step): + step = invocation_step.workflow_step progress.mark_step_outputs_delayed(step, why="executing pause step") - return None - def recover_mapping(self, step, step_invocations, progress): - if step_invocations: - step_invocation = step_invocations[0] - action = step_invocation.action + def recover_mapping(self, invocation_step, progress): + if invocation_step: + step = invocation_step.workflow_step + action = invocation_step.action if action: connection = step.input_connections_by_name["input"][0] replacement = progress.replacement_for_connection(connection) - progress.set_step_outputs(step, {'output': replacement}) + progress.set_step_outputs(invocation_step, {'output': replacement}) return elif action is False: raise CancelWorkflowEvaluation() @@ -790,7 +804,9 @@ class ToolModule(WorkflowModule): else: raise ToolMissingException("Tool %s missing. Cannot recover runtime state." % self.tool_id) - def execute(self, trans, progress, invocation, step): + def execute(self, trans, progress, invocation_step): + invocation = invocation_step.workflow_invocation + step = invocation_step.workflow_step tool = trans.app.toolbox.get_tool(step.tool_id, tool_version=step.tool_version) tool_state = step.state # Not strictly needed - but keep Tool state clean by stripping runtime @@ -855,6 +871,9 @@ class ToolModule(WorkflowModule): param_combinations.append(execution_state.inputs) + # Will be set if only a subset of required jobs have been scheduled and the + # workflow should be delayed. + partial_jobs = None try: execution_tracker = execute( trans=self.trans, @@ -862,50 +881,49 @@ class ToolModule(WorkflowModule): param_combinations=param_combinations, history=invocation.history, collection_info=collection_info, - workflow_invocation_uuid=invocation.uuid.hex + workflow_invocation_uuid=invocation.uuid.hex, + invocation_step=invocation_step, ) + except PartialJobExecution as p: + partial_jobs = p.jobs except ToolInputsNotReadyException: delayed_why = "tool [%s] inputs are not ready, this special tool requires inputs to be ready" % tool.id raise DelayedWorkflowEvaluation(why=delayed_why) - if collection_info: - step_outputs = dict(execution_tracker.implicit_collections) + if partial_jobs is None: + if collection_info: + step_outputs = dict(execution_tracker.implicit_collections) + else: + step_outputs = dict(execution_tracker.output_datasets) + step_outputs.update(execution_tracker.output_collections) + progress.set_step_outputs(invocation_step, step_outputs) + jobs = execution_tracker.successful_jobs else: - step_outputs = dict(execution_tracker.output_datasets) - step_outputs.update(execution_tracker.output_collections) - progress.set_step_outputs(step, step_outputs) - jobs = execution_tracker.successful_jobs + jobs = partial_jobs + for job in jobs: self._handle_post_job_actions(step, job, invocation.replacement_dict) + if execution_tracker.execution_errors: failed_count = len(execution_tracker.execution_errors) success_count = len(execution_tracker.successful_jobs) all_count = failed_count + success_count message = "Failed to create %d out of %s job(s) for workflow step." % (failed_count, all_count) raise Exception(message) - return jobs - def recover_mapping(self, step, step_invocations, progress): - # Grab a job representing this invocation - for normal workflows - # there will be just one job but if this step was mapped over there - # may be many. - job_0 = step_invocations[0].job + complete = partial_jobs is None + return jobs, complete + def recover_mapping(self, invocation_step, progress): outputs = {} - for job_output in job_0.output_datasets: - replacement_name = job_output.name - replacement_value = job_output.dataset - # If was a mapping step, grab the output mapped collection for - # replacement instead. - if replacement_value.hidden_beneath_collection_instance: - replacement_value = replacement_value.hidden_beneath_collection_instance - outputs[replacement_name] = replacement_value - for job_output_collection in job_0.output_dataset_collection_instances: - replacement_name = job_output_collection.name - replacement_value = job_output_collection.dataset_collection_instance - outputs[replacement_name] = replacement_value - progress.set_step_outputs(step, outputs) + for output_dataset_assoc in invocation_step.output_datasets: + outputs[output_dataset_assoc.output_name] = output_dataset_assoc.dataset + + for output_dataset_collection_assoc in invocation_step.output_dataset_collections: + outputs[output_dataset_collection_assoc.output_name] = output_dataset_collection_assoc.dataset_collection + + progress.set_step_outputs(invocation_step, outputs) def _find_collections_to_match(self, tool, progress, step): collections_to_match = matching.CollectionsToMatch() diff --git a/lib/galaxy/workflow/run.py b/lib/galaxy/workflow/run.py index 0f6fd041d04..75222d5a46d 100644 --- a/lib/galaxy/workflow/run.py +++ b/lib/galaxy/workflow/run.py @@ -162,24 +162,46 @@ class WorkflowInvoker(object): remaining_steps = self.progress.remaining_steps() delayed_steps = False - for step in remaining_steps: + for (step, workflow_invocation_step) in remaining_steps: step_delayed = False step_timer = ExecutionTimer() jobs = None try: self.__check_implicitly_dependent_steps(step) - # TODO: step may fail to invoke, do something about that. - jobs = self._invoke_step(step) - for job in (util.listify(jobs) or [None]): - # Record invocation + if not workflow_invocation_step: workflow_invocation_step = model.WorkflowInvocationStep() workflow_invocation_step.workflow_invocation = workflow_invocation workflow_invocation_step.workflow_step = step + workflow_invocation_step.state = 'new' + previously_executed_jobs_count = 0 + + workflow_invocation.steps.append(workflow_invocation_step) + else: + previously_executed_jobs_count = len(workflow_invocation_step.jobs) + + jobs_or_none = self._invoke_step(workflow_invocation_step) + if jobs_or_none: + jobs, complete = jobs_or_none + else: + jobs, complete = [], True + + for job in (util.listify(jobs) or []): + job_assoc = model.WorkflowInvocationStepJobAssociation() + job_assoc.index = previously_executed_jobs_count + job_assoc.workflow_invocation_step = workflow_invocation_step # Job may not be generated in this thread if bursting is enabled # https://github.com/galaxyproject/galaxy/issues/2259 - if job: - workflow_invocation_step.job_id = job.id + job_assoc.job_id = job.id + + previously_executed_jobs_count += 1 + + if not complete: + step_delayed = delayed_steps = True + workflow_invocation_step.state = 'ready' + self.progress.mark_step_outputs_delayed(step, why="Not all jobs scheduled for state.") + else: + workflow_invocation_step.state = 'scheduled' except modules.DelayedWorkflowEvaluation as de: step_delayed = delayed_steps = True self.progress.mark_step_outputs_delayed(step, why=de.why) @@ -218,15 +240,19 @@ class WorkflowInvoker(object): self.__check_implicitly_dependent_step(output_id) def __check_implicitly_dependent_step(self, output_id): - step_invocations = self.workflow_invocation.step_invocations_for_step_id(output_id) + step_invocation = self.workflow_invocation.step_invocation_for_step_id(output_id) # No steps created yet - have to delay evaluation. - if not step_invocations: + if not step_invocation: delayed_why = "depends on step [%s] but that step has not been invoked yet" % output_id raise modules.DelayedWorkflowEvaluation(why=delayed_why) - for step_invocation in step_invocations: - job = step_invocation.job + if step_invocation.state != 'scheduled': + delayed_why = "depends on step [%s] job has not finished scheduling yet" % output_id + raise modules.DelayedWorkflowEvaluation(delayed_why) + + for job_assoc in step_invocation.jobs: + job = job_assoc.job if job: # At least one job in incomplete. if not job.finished: @@ -241,9 +267,9 @@ class WorkflowInvoker(object): # pause steps. pass - def _invoke_step(self, step): - jobs = step.module.execute(self.trans, self.progress, self.workflow_invocation, step) - return jobs + def _invoke_step(self, invocation_step): + jobs_or_none = invocation_step.workflow_step.module.execute(self.trans, self.progress, invocation_step) + return jobs_or_none STEP_OUTPUT_DELAYED = object() @@ -256,11 +282,19 @@ class WorkflowProgress(object): self.module_injector = module_injector self.workflow_invocation = workflow_invocation self.inputs_by_step_id = inputs_by_step_id + self.jobs_per_scheduling_iteration = 1 + + @property + def maximum_jobs_to_schedule(self): + return 1 def remaining_steps(self): # Previously computed and persisted step states. step_states = self.workflow_invocation.step_states_by_step_id() steps = self.workflow_invocation.workflow.steps + + # TODO: Wouldn't a generator be much better here so we don't have to reason about + # steps we are no where near ready to schedule? remaining_steps = [] step_invocations_by_id = self.workflow_invocation.step_invocations_by_step_id() for step in steps: @@ -274,11 +308,11 @@ class WorkflowProgress(object): runtime_state = step_states[step_id].value step.state = step.module.decode_runtime_state(runtime_state) - invocation_steps = step_invocations_by_id.get(step_id, None) - if invocation_steps: - self._recover_mapping(step, invocation_steps) + invocation_step = step_invocations_by_id.get(step_id, None) + if invocation_step and invocation_step.state == 'scheduled': + self._recover_mapping(invocation_step) else: - remaining_steps.append(step) + remaining_steps.append((step, invocation_step)) return remaining_steps def replacement_for_tool_input(self, step, input, prefixed_name): @@ -340,7 +374,9 @@ class WorkflowProgress(object): output_name = workflow_output.output_name return self.outputs[step.id][output_name] - def set_outputs_for_input(self, step, outputs=None): + def set_outputs_for_input(self, invocation_step, outputs=None): + step = invocation_step.workflow_step + if outputs is None: outputs = {} @@ -352,10 +388,34 @@ class WorkflowProgress(object): raise ValueError(message) outputs['output'] = self.inputs_by_step_id[step_id] - self.set_step_outputs(step, outputs) + self.set_step_outputs(invocation_step, outputs) - def set_step_outputs(self, step, outputs): + def set_step_outputs(self, invocation_step, outputs, already_persisted=False): + step = invocation_step.workflow_step self.outputs[step.id] = outputs + if not already_persisted: + for output_name, output_object in outputs.items(): + if hasattr(output_object, "history_content_type"): + invocation_step.add_output(output_name, output_object) + else: + # This is a problem, this non-data, non-collection output + # won't be recovered on a subsequent workflow scheduling + # iteration. This seems to have been a pre-existing problem + # prior to #4584 though. + pass + for workflow_output in step.workflow_outputs: + output_name = workflow_output.output_name + if output_name not in outputs: + raise KeyError("Failed to find [%s] in step outputs [%s]" % (output_name, outputs)) + output = outputs[output_name] + self._record_workflow_output( + step, + workflow_output, + output=output, + ) + + def _record_workflow_output(self, step, workflow_output, output): + self.workflow_invocation.add_output(workflow_output, step, output) def mark_step_outputs_delayed(self, step, why=None): if why: @@ -414,11 +474,11 @@ class WorkflowProgress(object): self.module_injector, ) - def _recover_mapping(self, step, step_invocations): + def _recover_mapping(self, step_invocation): try: - step.module.recover_mapping(step, step_invocations, self) + step_invocation.workflow_step.module.recover_mapping(step_invocation, self) except modules.DelayedWorkflowEvaluation as de: - self.mark_step_outputs_delayed(step, de.why) + self.mark_step_outputs_delayed(step_invocation.workflow_step, de.why) __all__ = ('invoke', 'WorkflowRunConfig') diff --git a/test/api/test_workflows.py b/test/api/test_workflows.py index f875c2c9fa9..214692f1021 100644 --- a/test/api/test_workflows.py +++ b/test/api/test_workflows.py @@ -963,6 +963,77 @@ test_data: time.sleep(5) self.dataset_populator.wait_for_history(history_id, assert_ok=True) + def test_workflow_output_dataset(self): + history_id = self.dataset_populator.new_history() + summary = self._run_jobs(""" +class: GalaxyWorkflow +inputs: + - id: input1 +outputs: + - id: wf_output_1 + source: first_cat#out_file1 +steps: + - tool_id: cat1 + label: first_cat + state: + input1: + $link: input1 + +test_data: + input1: "hello world" +""", history_id=history_id) + workflow_id = summary.workflow_id + invocation_id = summary.invocation_id + invocation_response = self._get("workflows/%s/invocations/%s" % (workflow_id, invocation_id)) + self._assert_status_code_is(invocation_response, 200) + invocation = invocation_response.json() + self._assert_has_keys(invocation , "id", "outputs", "output_collections") + assert len(invocation["output_collections"]) == 0 + assert len(invocation["outputs"]) == 1 + output_content = self.dataset_populator.get_history_dataset_content(history_id, dataset_id=invocation["outputs"]["wf_output_1"]["id"]) + assert "hello world" == output_content.strip() + + def test_workflow_output_dataset_collection(self): + history_id = self.dataset_populator.new_history() + summary = self._run_jobs(""" +class: GalaxyWorkflow +inputs: + - id: input1 + type: data_collection_input + collection_type: list +outputs: + - id: wf_output_1 + source: first_cat#out_file1 +steps: + - tool_id: cat + label: first_cat + state: + input1: + $link: input1 +test_data: + input1: + type: list + name: the_dataset_list + elements: + - identifier: el1 + value: 1.fastq + type: File +""", history_id=history_id) + workflow_id = summary.workflow_id + invocation_id = summary.invocation_id + invocation_response = self._get("workflows/%s/invocations/%s" % (workflow_id, invocation_id)) + self._assert_status_code_is(invocation_response, 200) + invocation = invocation_response.json() + self._assert_has_keys(invocation , "id", "outputs", "output_collections") + assert len(invocation["output_collections"]) == 1 + assert len(invocation["outputs"]) == 0 + output_content = self.dataset_populator.get_history_collection_details(history_id, content_id=invocation["output_collections"]["wf_output_1"]["id"]) + self._assert_has_keys(output_content , "id", "elements") + elements = output_content["elements"] + assert len(elements) == 1 + elements0 = elements[0] + assert elements0["element_identifier"] == "el1" + @skip_without_tool("cat") def test_cancel_new_workflow_when_history_deleted(self): with self.dataset_populator.test_history() as history_id: diff --git a/test/base/populators.py b/test/base/populators.py index e382ace1ead..b64b57ff526 100644 --- a/test/base/populators.py +++ b/test/base/populators.py @@ -282,6 +282,8 @@ class BaseDatasetPopulator(object): # the last dataset in the history will be fetched. if "dataset_id" in kwds: history_content_id = kwds["dataset_id"] + elif "content_id" in kwds: + history_content_id = kwds["content_id"] elif "dataset" in kwds: history_content_id = kwds["dataset"]["id"] else: diff --git a/test/unit/workflows/test_workflow_progress.py b/test/unit/workflows/test_workflow_progress.py index 21cb2341414..3991cb393a4 100644 --- a/test/unit/workflows/test_workflow_progress.py +++ b/test/unit/workflows/test_workflow_progress.py @@ -72,13 +72,17 @@ class WorkflowProgressTestCase(unittest.TestCase): self.invocation, self.inputs_by_step_id, MockModuleInjector(self.progress) ) - def _set_previous_progress(self, outputs_dict): - for step_id, step_value in outputs_dict.items(): + def _set_previous_progress(self, outputs): + for i, (step_id, step_value) in enumerate(outputs): if step_value is not UNSCHEDULED_STEP: self.progress[step_id] = step_value workflow_invocation_step = model.WorkflowInvocationStep() workflow_invocation_step.workflow_step_id = step_id + workflow_invocation_step.state = 'scheduled' + workflow_invocation_step.workflow_step = self._step(i) + self.assertEqual(step_id, self._step(i).id) + # workflow_invocation_step.workflow_invocation = self.invocation self.invocation.steps.append(workflow_invocation_step) workflow_invocation_step_state = model.WorkflowRequestStepState() @@ -89,13 +93,21 @@ class WorkflowProgressTestCase(unittest.TestCase): def _step(self, index): return self.invocation.workflow.steps[index] + def _invocation_step(self, index): + if index < len(self.invocation.steps): + return self.invocation.steps[index] + else: + workflow_invocation_step = model.WorkflowInvocationStep() + workflow_invocation_step.workflow_step = self._step(index) + return workflow_invocation_step + def test_connect_data_input(self): self._setup_workflow(TEST_WORKFLOW_YAML) hda = model.HistoryDatasetAssociation() self.inputs_by_step_id = {100: hda} progress = self._new_workflow_progress() - progress.set_outputs_for_input(self._step(0)) + progress.set_outputs_for_input(self._invocation_step(0)) conn = model.WorkflowStepConnection() conn.output_name = "output" @@ -108,7 +120,7 @@ class WorkflowProgressTestCase(unittest.TestCase): self.inputs_by_step_id = {100: hda} progress = self._new_workflow_progress() - progress.set_outputs_for_input(self._step(0)) + progress.set_outputs_for_input(self._invocation_step(0)) replacement = progress.replacement_for_tool_input(self._step(2), MockInput(), "input1") assert replacement is hda @@ -118,7 +130,7 @@ class WorkflowProgressTestCase(unittest.TestCase): hda = model.HistoryDatasetAssociation() progress = self._new_workflow_progress() - progress.set_step_outputs(self._step(2), {"out1": hda}) + progress.set_step_outputs(self._invocation_step(2), {"out1": hda}) conn = model.WorkflowStepConnection() conn.output_name = "out1" @@ -128,17 +140,18 @@ class WorkflowProgressTestCase(unittest.TestCase): def test_remaining_steps_with_progress(self): self._setup_workflow(TEST_WORKFLOW_YAML) hda3 = model.HistoryDatasetAssociation() - self._set_previous_progress({ - 100: {"output": model.HistoryDatasetAssociation()}, - 101: {"output": model.HistoryDatasetAssociation()}, - 102: {"out_file1": hda3}, - 103: {"out_file1": model.HistoryDatasetAssociation()}, - 104: UNSCHEDULED_STEP, - }) + self._set_previous_progress([ + (100, {"output": model.HistoryDatasetAssociation()}), + (101, {"output": model.HistoryDatasetAssociation()}), + (102, {"out_file1": hda3}), + (103, {"out_file1": model.HistoryDatasetAssociation()}), + (104, UNSCHEDULED_STEP), + ]) progress = self._new_workflow_progress() steps = progress.remaining_steps() - assert len(steps) == 1 - assert steps[0] is self.invocation.workflow.steps[4] + assert len(steps) == 1, steps + step, invocation_step = steps[0] + assert step is self.invocation.workflow.steps[4] replacement = progress.replacement_for_tool_input(self._step(4), MockInput(), "input1") assert replacement is hda3 @@ -151,21 +164,27 @@ class WorkflowProgressTestCase(unittest.TestCase): def test_subworkflow_progress(self): self._setup_workflow(TEST_SUBWORKFLOW_YAML) hda = model.HistoryDatasetAssociation() - self._set_previous_progress({ - 100: {"output": hda}, - 101: UNSCHEDULED_STEP, - }) + self._set_previous_progress([ + (100, {"output": hda}), + (101, UNSCHEDULED_STEP), + ]) self.invocation.create_subworkflow_invocation_for_step( self.invocation.workflow.step_by_index(1) ) progress = self._new_workflow_progress() remaining_steps = progress.remaining_steps() - subworkflow_step = remaining_steps[0] + (subworkflow_step, subworkflow_invocation_step) = remaining_steps[0] subworkflow_progress = progress.subworkflow_progress(subworkflow_step) subworkflow = subworkflow_step.subworkflow assert subworkflow_progress.workflow_invocation.workflow == subworkflow + subworkflow_input_step = subworkflow.step_by_index(0) - subworkflow_progress.set_outputs_for_input(subworkflow_input_step) + subworkflow_invocation_step = model.WorkflowInvocationStep() + subworkflow_invocation_step.workflow_step_id = subworkflow_input_step.id + subworkflow_invocation_step.state = 'new' + subworkflow_invocation_step.workflow_step = subworkflow_input_step + + subworkflow_progress.set_outputs_for_input(subworkflow_invocation_step) subworkflow_cat_step = subworkflow.step_by_index(1) @@ -200,7 +219,7 @@ class MockModule(object): def decode_runtime_state(self, runtime_state): return True - def recover_mapping(self, step, step_invocations, progress): - step_id = step.id + def recover_mapping(self, invocation_step, progress): + step_id = invocation_step.workflow_step.id if step_id in self.progress: - progress.set_step_outputs(step, self.progress[step_id]) + progress.set_step_outputs(invocation_step, self.progress[step_id])