From d1cd4ab4d28c72f365852b75e2f2026c6b76ebbd Mon Sep 17 00:00:00 2001 From: John Chilton Date: Thu, 13 Nov 2014 13:47:38 -0500 Subject: [PATCH] Implement workflow scheduling 'plugin' framework. Models: Workflow invocations have been augmented with significantly more state - inputs, parameters, runtime step state, are all being tracked now. Workflow invocations have a state that can be changed over time, the UUIDs generated for workflow invocations in Pull Request #465 have to be persisted so they can be reused when scheduling new jobs for theworkflow invocation. Workflow invocation steps now have an action parameter for persisting state provided by users during the execution of the workflow (see forthcoming PauseModule for further details). Some initial elements of these model changes were based on model changes in Kyle Ellrott's Galaxy farm work (https://bitbucket.org/kellrott/galaxy-farm/branch/workflow_migrate). I made heavy modifications to the model to enforce referential integrity on parameter to workflow step mappings and made some cosmetic changes various other details. Scheduling Plugins: Used the pattern setup with dependency resolvers and job metrics to build a dynamic plugin infrastructure for defining workflow schedulers. I hesistate calling anything with only one implementation a plugin infrastructure, but I am confident enough that the combination of persisted workflow request combined with scheduler tag could be used to build a galaxy-farm plugin that would wait for another Galaxy instance to become available and it would pull the workflow down and This work piggy backs on Galaxy job handlers to have workflow scheduled in the background (i.e. during submission each workflow being scheduled in the background is assigned a unique job handler and only that job handler thread will process the workflow). It should be pretty easy to allow the definition of a new kind of handler - that is a workflow handler instead of a job handler if that is of interest. I will probably move a bunch of stuff that is happening in workflow/scheduling_manager.py more into the scheduler itself so that it can be more configurable and closer to a true plugin. API: There are a number of new API points here for flushing out dealing with workflow invocations (called usages in existing parlance). - POST /api/workflows/{encoded_workflow_id}/usage Schedule a worklfow to be run in the background and return just the workflow invocation information. RESTfully speaking this should be plural but the matching GET endpoint is likewise usage and not usages - so I am favoring consistency over RESTful correctness here. Also, likewise creating a 'usage' feel like odd - I would like to make all of the usage endpoints aliases to a more RESTfully correct invocations endpoints. The existing workflow run API endpoints still work and still work the way they use usually - but the output now includes all of the workflow invocation to_dict stuff as well as the list of outputs it initially used. Once everything is scheduled this way - that list of outputs is going to have to disappear but hopefully people can start using the invocation stuff now to help the transition. - DELETE /api/workflows/{workflow_id}/usage/{usage_id} Cancel a scheduled workflow invocation. - GET /api/workflows/{workflow_id}/usage/{usage_id}/steps/{step_id} Get information about a workflow invocation step. - PUT /api/workflows/{workflow_id}/usage/{usage_id}/steps/{step_id} Update a workflow invocation step - for ones with modifiable state. Extension point added to workflow modules to support this but it is unused by all existing worklfow modules. A subsequent PauseModule will use this to either continue or cancel a workflow invocation at a particular step. Modules: Workflow modules can now define new methods for dealing with recovering state and interacting with user requests. Testing: One can issue a workflow request by running the following test. ./run_tests.sh -api test/api/test_workflows.py:WorkflowsApiTestCase.test_workflow_request --- .hgignore | 1 + config/workflow_schedulers_conf.xml.sample | 14 ++ lib/galaxy/app.py | 9 + lib/galaxy/config.py | 1 + lib/galaxy/jobs/manager.py | 2 +- lib/galaxy/managers/workflows.py | 41 ++++ lib/galaxy/model/__init__.py | 136 +++++++++++- lib/galaxy/model/mapping.py | 86 +++++++- .../0123_add_workflow_request_tables.py | 144 +++++++++++++ lib/galaxy/webapps/galaxy/api/workflows.py | 120 ++++++++++- lib/galaxy/webapps/galaxy/buildapp.py | 5 + .../webapps/galaxy/controllers/workflow.py | 2 +- lib/galaxy/work/__init__.py | 0 lib/galaxy/work/context.py | 46 ++++ lib/galaxy/workflow/modules.py | 70 ++++++- lib/galaxy/workflow/run.py | 127 +++++++++-- lib/galaxy/workflow/run_request.py | 96 +++++++++ lib/galaxy/workflow/schedulers/__init__.py | 41 ++++ lib/galaxy/workflow/schedulers/core.py | 46 ++++ lib/galaxy/workflow/scheduling_manager.py | 197 ++++++++++++++++++ test/api/test_workflows.py | 30 ++- test/unit/test_galaxy_mapping.py | 49 +++++ test/unit/workflows/__init__.py | 0 23 files changed, 1224 insertions(+), 39 deletions(-) create mode 100644 config/workflow_schedulers_conf.xml.sample create mode 100644 lib/galaxy/model/migrate/versions/0123_add_workflow_request_tables.py create mode 100644 lib/galaxy/work/__init__.py create mode 100644 lib/galaxy/work/context.py create mode 100644 lib/galaxy/workflow/schedulers/__init__.py create mode 100644 lib/galaxy/workflow/schedulers/core.py create mode 100644 lib/galaxy/workflow/scheduling_manager.py create mode 100644 test/unit/workflows/__init__.py diff --git a/.hgignore b/.hgignore index b3ea5101292..b1cb3ca9553 100644 --- a/.hgignore +++ b/.hgignore @@ -67,6 +67,7 @@ data_manager_conf.xml shed_data_manager_conf.xml object_store_conf.xml job_metrics_conf.xml +workflow_schedulers_conf.xml config/* static/welcome.html.* static/welcome.html diff --git a/config/workflow_schedulers_conf.xml.sample b/config/workflow_schedulers_conf.xml.sample new file mode 100644 index 00000000000..c7572d9da6e --- /dev/null +++ b/config/workflow_schedulers_conf.xml.sample @@ -0,0 +1,14 @@ + + + + + + + + + diff --git a/lib/galaxy/app.py b/lib/galaxy/app.py index 9cda4879f70..69f27d3d233 100644 --- a/lib/galaxy/app.py +++ b/lib/galaxy/app.py @@ -142,6 +142,11 @@ class UniverseApplication( object, config.ConfiguresGalaxyMixin ): self.proxy_manager = ProxyManager( self.config ) # Initialize the external service types self.external_service_types = external_service_types.ExternalServiceTypesCollection( self.config.external_service_type_config_file, self.config.external_service_type_path, self ) + + from galaxy.workflow import scheduling_manager + # Must be initialized after job_config. + self.workflow_scheduling_manager = scheduling_manager.WorkflowSchedulingManager( self ) + self.model.engine.dispose() self.control_worker = GalaxyQueueWorker(self, galaxy.queues.control_queue_from_config(self.config), @@ -150,6 +155,7 @@ class UniverseApplication( object, config.ConfiguresGalaxyMixin ): self.control_worker.start() def shutdown( self ): + self.workflow_scheduling_manager.shutdown() self.job_manager.shutdown() self.object_store.shutdown() if self.heartbeat: @@ -171,3 +177,6 @@ class UniverseApplication( object, config.ConfiguresGalaxyMixin ): self.trace_logger = FluentTraceLogger( 'galaxy', self.config.fluent_host, self.config.fluent_port ) else: self.trace_logger = None + + def is_job_handler( self ): + return (self.config.track_jobs_in_database and self.job_config.is_handler(self.config.server_name)) or not self.config.track_jobs_in_database diff --git a/lib/galaxy/config.py b/lib/galaxy/config.py index 6f1d3194357..7415afc9347 100644 --- a/lib/galaxy/config.py +++ b/lib/galaxy/config.py @@ -139,6 +139,7 @@ class Configuration( object ): self.collect_outputs_from = [ x.strip() for x in kwargs.get( 'collect_outputs_from', 'new_file_path,job_working_directory' ).lower().split(',') ] self.template_path = resolve_path( kwargs.get( "template_path", "templates" ), self.root ) self.template_cache = resolve_path( kwargs.get( "template_cache_path", "database/compiled_templates" ), self.root ) + self.workflow_schedulers_config_file = resolve_path( kwargs.get( 'workflow_schedulers_config_file', 'config/workflow_schedulers_conf.xml' ), self.root ) self.local_job_queue_workers = int( kwargs.get( "local_job_queue_workers", "5" ) ) self.cluster_job_queue_workers = int( kwargs.get( "cluster_job_queue_workers", "3" ) ) self.job_queue_cleanup_interval = int( kwargs.get("job_queue_cleanup_interval", "5") ) diff --git a/lib/galaxy/jobs/manager.py b/lib/galaxy/jobs/manager.py index 20eeb8502c2..4cf56040f57 100644 --- a/lib/galaxy/jobs/manager.py +++ b/lib/galaxy/jobs/manager.py @@ -18,7 +18,7 @@ class JobManager( object ): """ def __init__( self, app ): self.app = app - if (self.app.config.track_jobs_in_database and self.app.job_config.is_handler(self.app.config.server_name)) or not self.app.config.track_jobs_in_database: + if self.app.is_job_handler(): log.debug("Starting job handler") self.job_handler = handler.JobHandler( app ) self.job_queue = self.job_handler.job_queue diff --git a/lib/galaxy/managers/workflows.py b/lib/galaxy/managers/workflows.py index af762780253..5d0246a3ba9 100644 --- a/lib/galaxy/managers/workflows.py +++ b/lib/galaxy/managers/workflows.py @@ -1,5 +1,6 @@ from galaxy import model from galaxy import exceptions +from galaxy.workflow import modules class WorkflowsManager( object ): @@ -49,6 +50,46 @@ class WorkflowsManager( object ): self.check_security( trans, workflow_invocation, check_ownership=True, check_accessible=False ) return workflow_invocation + def cancel_invocation( self, trans, decoded_invocation_id ): + workflow_invocation = self.get_invocation( trans, decoded_invocation_id ) + cancelled = workflow_invocation.cancel() + + if cancelled: + trans.sa_session.add( workflow_invocation ) + trans.sa_session.flush() + else: + # TODO: More specific exception? + raise exceptions.MessageException( "Cannot cancel an inactive workflow invocation." ) + + return workflow_invocation + + def get_invocation_step( self, trans, decoded_workflow_invocation_step_id ): + try: + workflow_invocation_step = trans.sa_session.query( + model.WorkflowInvocationStep + ).get( decoded_workflow_invocation_step_id ) + except Exception: + raise exceptions.ObjectNotFound() + self.check_security( trans, workflow_invocation_step.workflow_invocation, check_ownership=True, check_accessible=False ) + return workflow_invocation_step + + def update_invocation_step( self, trans, decoded_workflow_invocation_step_id, action ): + if action is None: + raise exceptions.RequestParameterMissingException( "Updating workflow invocation step requires an action parameter. " ) + + workflow_invocation_step = self.get_invocation_step( trans, decoded_workflow_invocation_step_id ) + workflow_invocation = workflow_invocation_step.workflow_invocation + if not workflow_invocation.active: + raise exceptions.RequestParameterInvalidException( "Attempting to modify the state of an completed workflow invocation." ) + + step = workflow_invocation_step.workflow_step + module = modules.module_factory.from_workflow_step( trans, step ) + performed_action = module.do_invocation_step_action( step, action ) + workflow_invocation_step.action = performed_action + trans.sa_session.add( workflow_invocation_step ) + trans.sa_session.flush() + return workflow_invocation_step + def build_invocations_query( self, trans, decoded_stored_workflow_id ): try: stored_workflow = trans.sa_session.query( diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py index 97328fa77c1..f124ccff1a8 100644 --- a/lib/galaxy/model/__init__.py +++ b/lib/galaxy/model/__init__.py @@ -39,6 +39,7 @@ from galaxy.web.framework.helpers import to_unicode from galaxy.web.form_builder import (AddressField, CheckboxField, HistoryField, PasswordField, SelectField, TextArea, TextField, WorkflowField, WorkflowMappingField) +from galaxy.model.orm import and_, or_ from sqlalchemy.orm import object_session from sqlalchemy.orm import joinedload from sqlalchemy.sql.expression import func @@ -3096,8 +3097,74 @@ class StoredWorkflowMenuEntry( object ): class WorkflowInvocation( object, Dictifiable ): - dict_collection_visible_keys = ( 'id', 'update_time', 'workflow_id' ) - dict_element_visible_keys = ( 'id', 'update_time', 'workflow_id' ) + dict_collection_visible_keys = ( 'id', 'update_time', 'workflow_id', 'history_id', 'uuid', 'state' ) + dict_element_visible_keys = ( 'id', 'update_time', 'workflow_id', 'history_id', 'uuid', 'state' ) + states = Bunch( + NEW='new', # Brand new workflow invocation... maybe this should be same as READY + READY='ready', # Workflow ready for another iteration of scheduling. + SCHEDULED='scheduled', # Workflow has been scheduled. + CANCELLED='cancelled', + FAILED='failed', + ) + + @property + def active( self ): + """ Indicates the workflow invocation is somehow active - and in + particular valid actions may be performed on its + ``WorkflowInvocationStep``s. + """ + states = WorkflowInvocation.states + return self.state in [ states.NEW, states.READY ] + + def cancel( self ): + if not self.active: + return False + else: + self.state = WorkflowInvocation.states.CANCELLED + return True + + def fail( self ): + self.state = WorkflowInvocation.states.FAILED + + def step_states_by_step_id( self ): + step_states = {} + for step_state in self.step_states: + step_id = step_state.workflow_step_id + step_states[ step_id ] = step_state + return step_states + + def step_invocations_by_step_id( self ): + 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 ) + return step_invocations + + @staticmethod + def poll_active_workflow_ids( + sa_session, + scheduler=None, + handler=None + ): + and_conditions = [ + or_( + WorkflowInvocation.state == WorkflowInvocation.states.NEW, + WorkflowInvocation.state == WorkflowInvocation.states.READY + ), + ] + if scheduler is not None: + and_conditions.append( WorkflowInvocation.scheduler == scheduler ) + if handler is not None: + and_conditions.append( WorkflowInvocation.handler == handler ) + + query = sa_session.query( + WorkflowInvocation + ).filter( and_( *and_conditions ) ) + # Immediately just load all ids into memory so time slicing logic + # is relatively intutitive. + return map( lambda wi: wi.id, query.all() ) def to_dict( self, view='collection', value_mapper=None ): rval = super( WorkflowInvocation, self ).to_dict( view=view, value_mapper=value_mapper ) @@ -3123,14 +3190,63 @@ class WorkflowInvocation( object, Dictifiable ): class WorkflowInvocationStep( object, Dictifiable ): - dict_collection_visible_keys = ( 'id', 'update_time', 'job_id', 'workflow_step_id' ) - dict_element_visible_keys = ( 'id', 'update_time', 'job_id', 'workflow_step_id' ) + 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' ) 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 return rval + +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' ] + + def to_dict( self, view='collection', value_mapper=None ): + rval = super( WorkflowRequest, self ).to_dict( view=view, value_mapper=value_mapper ) + return rval + + +class WorkflowRequestInputParameter(object, Dictifiable): + """ Workflow-related parameters not tied to steps or inputs. + """ + dict_collection_visible_keys = ['id', 'name', 'value', 'type'] + types = Bunch( + REPLACEMENT_PARAMETERS='replacements', + META_PARAMETERS='meta', # + ) + + def __init__( self, name=None, value=None, type=None ): + self.name = name + self.value = value + self.type = type + + +class WorkflowRequestStepState(object, Dictifiable): + """ Workflow step value parameters. + """ + dict_collection_visible_keys = ['id', 'name', 'value', 'workflow_step_id'] + + def __init__( self, workflow_step=None, name=None, value=None ): + self.workflow_step = workflow_step + self.name = name + self.value = value + self.type = type + + +class WorkflowRequestToInputDatasetAssociation(object, Dictifiable): + """ Workflow step input dataset parameters. + """ + dict_collection_visible_keys = ['id', 'workflow_invocation_id', 'workflow_step_id', 'dataset_id', 'name' ] + + +class WorkflowRequestToInputDatasetCollectionAssociation(object, Dictifiable): + """ Workflow step input dataset collection parameters. + """ + dict_collection_visible_keys = ['id', 'workflow_invocation_id', 'workflow_step_id', 'dataset_collection_id', 'name' ] + + class MetadataFile( object ): def __init__( self, dataset=None, name=None ): @@ -4038,6 +4154,18 @@ class ToolTagAssociation( ItemTagAssociation ): self.value = None self.user_value = None + +class WorkRequestTagAssociation( ItemTagAssociation ): + def __init__( self, id=None, user=None, workflow_request_id=None, tag_id=None, user_tname=None, value=None ): + self.id = id + self.user = user + self.workflow_request_id = workflow_request_id + self.tag_id = tag_id + self.user_tname = user_tname + self.value = None + self.user_value = None + + # Item annotation classes. class HistoryAnnotationAssociation( object ): diff --git a/lib/galaxy/model/mapping.py b/lib/galaxy/model/mapping.py index 8f7ca819444..d96f5b41f30 100644 --- a/lib/galaxy/model/mapping.py +++ b/lib/galaxy/model/mapping.py @@ -710,6 +710,46 @@ model.WorkflowStep.table = Table( "workflow_step", metadata, ## Column( "input_connections", JSONType ) ) + +model.WorkflowRequestStepState.table = Table( + "workflow_request_step_states", metadata, + Column( "id", Integer, primary_key=True ), + Column( "workflow_invocation_id", Integer, ForeignKey("workflow_invocation.id", onupdate="CASCADE", ondelete="CASCADE" )), + Column( "workflow_step_id", Integer, ForeignKey("workflow_step.id" )), + Column( "value", JSONType ), +) + + +model.WorkflowRequestInputParameter.table = Table( + "workflow_request_input_parameters", metadata, + Column( "id", Integer, primary_key=True ), + Column( "workflow_invocation_id", Integer, ForeignKey("workflow_invocation.id", onupdate="CASCADE", ondelete="CASCADE" )), + Column( "name", Unicode(255) ), + Column( "value", TEXT ), + Column( "type", Unicode(255) ), +) + + +model.WorkflowRequestToInputDatasetAssociation.table = Table( + "workflow_request_to_input_dataset", metadata, + Column( "id", Integer, primary_key=True ), + Column( "name", String(255) ), + 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 ), +) + + +model.WorkflowRequestToInputDatasetCollectionAssociation.table = Table( + "workflow_request_to_input_collection_dataset", metadata, + Column( "id", Integer, primary_key=True ), + Column( "name", String(255) ), + 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 ), +) + + model.WorkflowStepConnection.table = Table( "workflow_step_connection", metadata, Column( "id", Integer, primary_key=True ), Column( "output_step_id", Integer, ForeignKey( "workflow_step.id" ), index=True ), @@ -728,8 +768,13 @@ model.WorkflowInvocation.table = Table( "workflow_invocation", metadata, Column( "id", Integer, primary_key=True ), Column( "create_time", DateTime, default=now ), Column( "update_time", DateTime, default=now, onupdate=now ), - Column( "workflow_id", Integer, ForeignKey( "workflow.id" ), index=True, nullable=False ) - ) + Column( "workflow_id", Integer, ForeignKey( "workflow.id" ), index=True, nullable=False ), + Column( "state", TrimmedString( 64 ), index=True ), + Column( "scheduler", TrimmedString( 255 ), index=True ), + Column( "handler", TrimmedString( 255 ), index=True ), + Column( 'uuid', UUIDType() ), + Column( "history_id", Integer, ForeignKey( "history.id" ), index=True ), +) model.WorkflowInvocationStep.table = Table( "workflow_invocation_step", metadata, Column( "id", Integer, primary_key=True ), @@ -737,8 +782,9 @@ model.WorkflowInvocationStep.table = Table( "workflow_invocation_step", metadata 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( "job_id", Integer, ForeignKey( "job.id" ), index=True, nullable=True ), + Column( "action", JSONType, nullable=True ), +) model.StoredWorkflowUserShareAssociation.table = Table( "stored_workflow_user_share_connection", metadata, Column( "id", Integer, primary_key=True ), @@ -1968,6 +2014,11 @@ mapper( model.StoredWorkflowMenuEntry, model.StoredWorkflowMenuEntry.table, mapper( model.WorkflowInvocation, model.WorkflowInvocation.table, properties=dict( + history=relation( model.History ), + input_parameters=relation( model.WorkflowRequestInputParameter ), + step_states=relation( model.WorkflowRequestStepState ), + input_datasets=relation( model.WorkflowRequestToInputDatasetAssociation ), + input_dataset_collections=relation( model.WorkflowRequestToInputDatasetCollectionAssociation ), steps=relation( model.WorkflowInvocationStep, backref='workflow_invocation', lazy=False ), workflow=relation( model.Workflow ) ) ) @@ -1976,6 +2027,33 @@ mapper( model.WorkflowInvocationStep, model.WorkflowInvocationStep.table, workflow_step = relation( model.WorkflowStep ), job = relation( model.Job, backref=backref( 'workflow_invocation_step', uselist=False ) ) ) ) +simple_mapping( + model.WorkflowRequestInputParameter, + workflow_invocation=relation( model.WorkflowInvocation ), +) + +simple_mapping( + model.WorkflowRequestStepState, + workflow_invocation=relation( model.WorkflowInvocation ), + workflow_step=relation( model.WorkflowStep ), +) + +simple_mapping( + model.WorkflowRequestToInputDatasetAssociation, + workflow_invocation=relation( model.WorkflowInvocation ), + workflow_step=relation( model.WorkflowStep ), + dataset=relation( model.HistoryDatasetAssociation ), +) + + +simple_mapping( + model.WorkflowRequestToInputDatasetCollectionAssociation, + workflow_invocation=relation( model.WorkflowInvocation ), + workflow_step=relation( model.WorkflowStep ), + dataset_collection=relation( model.HistoryDatasetCollectionAssociation ), +) + + mapper( model.MetadataFile, model.MetadataFile.table, properties=dict( history_dataset=relation( model.HistoryDatasetAssociation ), library_dataset=relation( model.LibraryDatasetDatasetAssociation ) ) ) diff --git a/lib/galaxy/model/migrate/versions/0123_add_workflow_request_tables.py b/lib/galaxy/model/migrate/versions/0123_add_workflow_request_tables.py new file mode 100644 index 00000000000..653220fad22 --- /dev/null +++ b/lib/galaxy/model/migrate/versions/0123_add_workflow_request_tables.py @@ -0,0 +1,144 @@ +""" +Migration script for workflow request tables. +""" +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() + + +WorkflowRequestInputParameter_table = Table( + "workflow_request_input_parameters", metadata, + Column( "id", Integer, primary_key=True ), + Column( "workflow_invocation_id", Integer, ForeignKey("workflow_invocation.id", onupdate="CASCADE", ondelete="CASCADE" )), + Column( "name", Unicode(255) ), + Column( "type", Unicode(255) ), + Column( "value", TEXT ), +) + + +WorkflowRequestStepState_table = Table( + "workflow_request_step_states", metadata, + Column( "id", Integer, primary_key=True ), + Column( "workflow_invocation_id", Integer, ForeignKey("workflow_invocation.id", onupdate="CASCADE", ondelete="CASCADE" )), + Column( "workflow_step_id", Integer, ForeignKey("workflow_step.id" )), + Column( "value", JSONType ), +) + + +WorkflowRequestToInputDatasetAssociation_table = Table( + "workflow_request_to_input_dataset", metadata, + Column( "id", Integer, primary_key=True ), + Column( "name", String(255) ), + 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 ), +) + + +WorkflowRequestToInputDatasetCollectionAssociation_table = Table( + "workflow_request_to_input_collection_dataset", metadata, + Column( "id", Integer, primary_key=True ), + Column( "name", String(255) ), + 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 ), +) + + +TABLES = [ + WorkflowRequestInputParameter_table, + WorkflowRequestStepState_table, + WorkflowRequestToInputDatasetAssociation_table, + WorkflowRequestToInputDatasetCollectionAssociation_table, +] + + +def upgrade(migrate_engine): + metadata.bind = migrate_engine + print __doc__ + metadata.reflect() + + for table in TABLES: + __create(table) + + History_column = Column( "history_id", Integer, ForeignKey( "history.id" ), nullable=True ) + State_column = Column( "state", TrimmedString( 64 ) ) + + # TODO: Handle indexes correctly + SchedulerId_column = Column( "scheduler", TrimmedString(255) ) + HandlerId_column = Column( "handler", TrimmedString(255) ) + WorkflowUUID_column = Column( "uuid", UUIDType, nullable=True ) + __add_column( History_column, "workflow_invocation", metadata ) + __add_column( State_column, "workflow_invocation", metadata ) + __add_column( SchedulerId_column, "workflow_invocation", metadata, index_nane="id_workflow_invocation_scheduler" ) + __add_column( HandlerId_column, "workflow_invocation", metadata, index_name="id_workflow_invocation_handler" ) + __add_column( WorkflowUUID_column, "workflow_invocation", metadata ) + + # All previous invocations have been scheduled... + cmd = "UPDATE workflow_invocation SET state = 'scheduled'" + try: + migrate_engine.execute( cmd ) + except Exception, e: + log.debug( "failed to update past workflow invocation states: %s" % ( str( e ) ) ) + + WorkflowInvocationStepAction_column = Column( "action", JSONType, nullable=True ) + __add_column( WorkflowInvocationStepAction_column, "workflow_invocation_step", metadata ) + + +def downgrade(migrate_engine): + metadata.bind = migrate_engine + metadata.reflect() + + for table in TABLES: + __drop(table) + + __drop_column( "state", "workflow_invocation", metadata ) + __drop_column( "scheduler_id", "workflow_invocation", metadata ) + __drop_column( "uuid", "workflow_invocation", metadata ) + __drop_column( "history_id", "workflow_invocation", metadata ) + __drop_column( "handler_id", "workflow_invocation", metadata ) + __drop_column( "action", "workflow_invocation_step", 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 column 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 ) + + +def __create(table): + try: + table.create() + except Exception as e: + print str(e) + log.exception("Creating %s table failed: %s" % (table.name, str( e ) ) ) + + +def __drop(table): + try: + table.drop() + except Exception as e: + print str(e) + log.exception("Dropping %s table failed: %s" % (table.name, str( e ) ) ) diff --git a/lib/galaxy/webapps/galaxy/api/workflows.py b/lib/galaxy/webapps/galaxy/api/workflows.py index 8d2e27cfe97..588107d9e00 100644 --- a/lib/galaxy/webapps/galaxy/api/workflows.py +++ b/lib/galaxy/webapps/galaxy/api/workflows.py @@ -16,7 +16,7 @@ from galaxy.web.base.controller import BaseAPIController, url_for, UsesStoredWor from galaxy.web.base.controller import UsesHistoryMixin from galaxy.web.base.controller import SharableMixin from galaxy.workflow.extract import extract_workflow -from galaxy.workflow.run import invoke +from galaxy.workflow.run import invoke, queue_invoke from galaxy.workflow.run_request import build_workflow_run_config log = logging.getLogger(__name__) @@ -224,7 +224,7 @@ class WorkflowsAPIController(BaseAPIController, UsesStoredWorkflowMixin, UsesHis # invoke may throw MessageExceptions on tool erors, failure # to match up inputs, etc... - outputs = invoke( + outputs, invocation = invoke( trans=trans, workflow=workflow, workflow_run_config=run_config, @@ -235,14 +235,19 @@ class WorkflowsAPIController(BaseAPIController, UsesStoredWorkflowMixin, UsesHis # Build legacy output - should probably include more information from # outputs. rval = {} - rval['history'] = trans.security.encode_id(history.id) + rval['history'] = trans.security.encode_id( history.id ) rval['outputs'] = [] for step in workflow.steps: if step.type == 'tool' or step.type is None: for v in outputs[ step.id ].itervalues(): rval[ 'outputs' ].append( trans.security.encode_id( v.id ) ) - return rval + # Newer version of this API just returns the invocation as a dict, to + # facilitate migration - produce the newer style response and blend in + # the older information. + invocation_response = self.__encode_invocation( trans, invocation ) + invocation_response.update( rval ) + return invocation_response @expose_api def workflow_dict( self, trans, workflow_id, **kwd ): @@ -369,6 +374,32 @@ class WorkflowsAPIController(BaseAPIController, UsesStoredWorkflowMixin, UsesHis item['url'] = url_for('workflow', id=encoded_id) return item + @expose_api + def workflow_request( self, trans, workflow_id, payload, **kwd ): + """ + POST /api/workflows/{encoded_workflow_id}/usage + + Schedule the workflow specified by `workflow_id` to run. + """ + # /usage is awkward in this context but is consistent with the rest of + # this module. Would prefer to redo it all to use /invocation(s). + # Get workflow + accessibility check. + stored_workflow = self.__get_stored_accessible_workflow( trans, workflow_id ) + workflow = stored_workflow.latest_workflow + + run_config = build_workflow_run_config( trans, workflow, payload ) + workflow_scheduler_id = payload.get( "scheduler", None ) + # TODO: workflow scheduler hints + work_request_params = dict( scheduler=workflow_scheduler_id ) + + workflow_invocation = queue_invoke( + trans=trans, + workflow=workflow, + workflow_run_config=run_config, + request_params=work_request_params + ) + return self.encode_all_ids( trans, workflow_invocation.to_dict(), recursive=True ) + @expose_api def workflow_usage(self, trans, workflow_id, **kwd): """ @@ -407,6 +438,87 @@ class WorkflowsAPIController(BaseAPIController, UsesStoredWorkflowMixin, UsesHis return self.__encode_invocation( trans, workflow_invocation ) return None + @expose_api + def cancel_workflow_invocation(self, trans, workflow_id, usage_id, **kwd): + """ + DELETE /api/workflows/{workflow_id}/usage/{usage_id} + Cancel the specified workflow invocation. + + :param workflow_id: the workflow id (required) + :type workflow_id: str + + :param usage_id: the usage id (required) + :type usage_id: str + + :raises: exceptions.MessageException, exceptions.ObjectNotFound + """ + decoded_workflow_invocation_id = self.__decode_id( trans, usage_id ) + workflow_invocation = self.workflow_manager.cancel_invocation( trans, decoded_workflow_invocation_id ) + return self.__encode_invocation( trans, workflow_invocation ) + + @expose_api + def workflow_invocation_step(self, trans, workflow_id, usage_id, step_id, **kwd): + """ + GET /api/workflows/{workflow_id}/usage/{usage_id}/steps/{step_id} + + :param workflow_id: the workflow id (required) + :type workflow_id: str + + :param usage_id: the usage id (required) + :type usage_id: str + + :param step_id: encoded id of the WorkflowInvocationStep (required) + :type step_id: str + + :param payload: payload containing update action information + for running workflow. + + :raises: exceptions.MessageException, exceptions.ObjectNotFound + """ + decoded_invocation_step_id = self.__decode_id( trans, step_id ) + invocation_step = self.workflow_manager.get_invocation_step( + trans, + decoded_invocation_step_id + ) + return self.__encode_invocation_step( trans, invocation_step ) + + @expose_api + def workflow_invocation_step_update(self, trans, workflow_id, usage_id, step_id, payload, **kwd): + """ + PUT /api/workflows/{workflow_id}/usage/{usage_id}/steps/{step_id} + Update state of running workflow step invocation - still very nebulous + but this would be for stuff like confirming paused steps can proceed + etc.... + + + :param workflow_id: the workflow id (required) + :type workflow_id: str + + :param usage_id: the usage id (required) + :type usage_id: str + + :param step_id: encoded id of the WorkflowInvocationStep (required) + :type step_id: str + + :raises: exceptions.MessageException, exceptions.ObjectNotFound + """ + decoded_invocation_step_id = self.__decode_id( trans, step_id ) + action = payload.get( "action", None ) + + invocation_step = self.workflow_manager.update_invocation_step( + trans, + decoded_invocation_step_id, + action=action, + ) + return self.__encode_invocation_step( trans, invocation_step ) + + def __encode_invocation_step( self, trans, invocation_step ): + return self.encode_all_ids( + trans, + invocation_step.to_dict( 'element' ), + True + ) + def __get_stored_accessible_workflow( self, trans, workflow_id ): stored_workflow = self.__get_stored_workflow( trans, workflow_id ) diff --git a/lib/galaxy/webapps/galaxy/buildapp.py b/lib/galaxy/webapps/galaxy/buildapp.py index 5de6f2557d1..661c338be42 100644 --- a/lib/galaxy/webapps/galaxy/buildapp.py +++ b/lib/galaxy/webapps/galaxy/buildapp.py @@ -227,7 +227,12 @@ def populate_api_routes( webapp, app ): webapp.mapper.connect( 'import_shared_workflow_deprecated', '/api/workflows/import', controller='workflows', action='import_shared_workflow_deprecated', conditions=dict( method=['POST'] ) ) webapp.mapper.connect( 'workflow_usage', '/api/workflows/{workflow_id}/usage', controller='workflows', action='workflow_usage', conditions=dict(method=['GET'])) webapp.mapper.connect( 'workflow_usage_contents', '/api/workflows/{workflow_id}/usage/{usage_id}', controller='workflows', action='workflow_usage_contents', conditions=dict(method=['GET'])) + webapp.mapper.connect( 'cancel_workflow_invocation', '/api/workflows/{workflow_id}/usage/{usage_id}', controller='workflows', action='cancel_workflow_invocation', conditions=dict(method=['DELETE'])) + webapp.mapper.connect( 'workflow_invocation_step', '/api/workflows/{workflow_id}/usage/{usage_id}/steps/{step_id}', controller='workflows', action='workflow_invocation_step', conditions=dict(method=['GET'])) + webapp.mapper.connect( 'workflow_invocation_step_update', '/api/workflows/{workflow_id}/usage/{usage_id}/steps/{step_id}', controller='workflows', action='workflow_invocation_step_update', conditions=dict(method=['PUT'])) + + webapp.mapper.connect( 'workflow_request', '/api/workflows/{workflow_id}/usage', controller='workflows', action='workflow_request', conditions=dict( method=['POST'] ) ) # ============================ # ===== AUTHENTICATE API ===== # ============================ diff --git a/lib/galaxy/webapps/galaxy/controllers/workflow.py b/lib/galaxy/webapps/galaxy/controllers/workflow.py index cb9586c9a41..1ceb9ce915f 100644 --- a/lib/galaxy/webapps/galaxy/controllers/workflow.py +++ b/lib/galaxy/webapps/galaxy/controllers/workflow.py @@ -1295,7 +1295,7 @@ class WorkflowController( BaseUIController, SharableMixin, UsesStoredWorkflowMix copy_inputs_to_history=new_history is not None ) - outputs = invoke( + outputs, invocation = invoke( trans=trans, workflow=workflow, workflow_run_config=run_config diff --git a/lib/galaxy/work/__init__.py b/lib/galaxy/work/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/lib/galaxy/work/context.py b/lib/galaxy/work/context.py new file mode 100644 index 00000000000..47782440b00 --- /dev/null +++ b/lib/galaxy/work/context.py @@ -0,0 +1,46 @@ +from galaxy.managers.context import ( + ProvidesAppContext, + ProvidesUserContext, + ProvidesHistoryContext +) + + +class WorkRequestContext( ProvidesAppContext, ProvidesUserContext, ProvidesHistoryContext ): + """ Stripped down implementation of Galaxy web transaction god object for + work request handling outside of web threads - uses mix-ins shared with + GalaxyWebTransaction to provide app, user, and history context convience + methods - but nothing related to HTTP handling, mako views, etc.... + + Things that only need app shouldn't be consuming trans - but there is a + need for actions potentially tied to users and histories and hopefully + this can define that stripped down interface providing access to user and + history information - but not dealing with web request and response + objects. + """ + + def __init__( self, app, user=None, history=None ): + self.app = app + self.security = app.security + self.__user = user + self.__history = history + self.api_inherit_admin = False + + def get_history( self, create=False ): + if create: + raise NotImplementedError( "Cannot create histories from a work request context." ) + return self.__history + + def set_history( self ): + raise NotImplementedError( "Cannot change histories from a work request context." ) + + history = property( get_history, set_history ) + + def get_user( self ): + """Return the current user if logged in or None.""" + return self.__user + + def set_user( self, user ): + """Set the current user.""" + raise NotImplementedError( "Cannot change users from a work request context." ) + + user = property( get_user, set_user ) diff --git a/lib/galaxy/workflow/modules.py b/lib/galaxy/workflow/modules.py index df87c85285e..77943060387 100644 --- a/lib/galaxy/workflow/modules.py +++ b/lib/galaxy/workflow/modules.py @@ -171,6 +171,21 @@ class WorkflowModule( object ): """ raise TypeError( "Abstract method" ) + def do_invocation_step_action( self, step, action ): + """ Update or set the workflow invocation state action - generic + extension point meant to allows users to interact with interactive + workflow modules. The action object returned from this method will + be attached to the WorkflowInvocationStep and be available the next + time the workflow scheduler visits the workflow. + """ + raise exceptions.RequestParameterInvalidException( "Attempting to perform invocation step action on module that does not support actions." ) + + def recover_mapping( self, step, step_invocations, progress ): + """ Re-populate progress object with information about connections + from previously executed steps recorded via step_invocations. + """ + raise TypeError( "Abstract method" ) + class InputModule( WorkflowModule ): @@ -240,6 +255,19 @@ class InputModule( WorkflowModule ): state.inputs = dict( input=None ) return state + def recover_runtime_state( self, runtime_state ): + """ Take secure runtime state from persisted invocation and convert it + into a DefaultToolState object for use during workflow invocation. + """ + fake_tool = Bunch( inputs=self.get_runtime_inputs() ) + state = galaxy.tools.DefaultToolState() + state.decode( runtime_state, fake_tool, self.trans.app, secure=False ) + return state + + def normalize_runtime_state( self, runtime_state ): + fake_tool = Bunch( inputs=self.get_runtime_inputs() ) + return runtime_state.encode( fake_tool, self.trans.app, secure=False ) + def encode_runtime_state( self, trans, state ): fake_tool = Bunch( inputs=self.get_runtime_inputs() ) return state.encode( fake_tool, trans.app ) @@ -291,6 +319,9 @@ class InputModule( WorkflowModule ): progress.set_outputs_for_input( step, step_outputs ) return job + def recover_mapping( self, step, step_invocations, progress ): + progress.set_outputs_for_input( step ) + class InputDataModule( InputModule ): type = "data_input" @@ -439,6 +470,18 @@ class ToolModule( WorkflowModule ): ) self.state.inputs = self.tool.params_from_strings( state, app, **params_from_kwds ) + def recover_runtime_state( self, runtime_state ): + """ Take secure runtime state from persisted invocation and convert it + into a DefaultToolState object for use during workflow invocation. + """ + state = galaxy.tools.DefaultToolState() + app = self.trans.app + state.decode( runtime_state, self.tool, app, secure=False ) + return state + + def normalize_runtime_state( self, runtime_state ): + return runtime_state.encode( self.tool, self.trans.app, secure=False ) + @classmethod def __get_tool_version( cls, trans, tool_id ): # Return a ToolVersion if one exists for tool_id. @@ -664,7 +707,7 @@ class ToolModule( WorkflowModule ): param_combinations=param_combinations, history=invocation.history, collection_info=collection_info, - workflow_invocation_uuid=invocation.uuid + workflow_invocation_uuid=invocation.uuid.hex ) if collection_info: step_outputs = dict( execution_tracker.created_collections ) @@ -731,6 +774,23 @@ class ToolModule( WorkflowModule ): visit_input_values( self.tool.inputs, self.state.inputs, callback ) + 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 + + 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 + progress.set_step_outputs( step, outputs ) + class WorkflowModuleFactory( object ): @@ -796,6 +856,14 @@ class MissingToolException( Exception ): module is missing. """ +class DelayedWorkflowEvaluation(Exception): + pass + + +class CancelWorkflowEvaluation(Exception): + pass + + class WorkflowModuleInjector(object): """ Injects workflow step objects from the ORM with appropriate module and module generated/influenced state. """ diff --git a/lib/galaxy/workflow/run.py b/lib/galaxy/workflow/run.py index 679edcf9923..98fcc627659 100644 --- a/lib/galaxy/workflow/run.py +++ b/lib/galaxy/workflow/run.py @@ -6,54 +6,110 @@ from galaxy import util from galaxy.util.odict import odict from galaxy.workflow import modules from galaxy.workflow.run_request import WorkflowRunConfig +from galaxy.workflow.run_request import workflow_run_config_to_request import logging log = logging.getLogger( __name__ ) -def invoke( trans, workflow, workflow_run_config, populate_state=False ): +def invoke( trans, workflow, workflow_run_config, workflow_invocation=None, populate_state=False ): """ Run the supplied workflow in the supplied target_history. """ if populate_state: modules.populate_module_and_state( trans, workflow, workflow_run_config.param_map ) - return WorkflowInvoker( + invoker = WorkflowInvoker( trans, workflow, workflow_run_config, - ).invoke() + workflow_invocation=workflow_invocation, + ) + try: + outputs = invoker.invoke() + except modules.CancelWorkflowEvaluation: + if workflow_invocation: + if workflow_invocation.cancel(): + trans.sa_session.add( workflow_invocation ) + outputs = [] + except Exception: + log.exception("Failed to execute scheduled workflow.") + if workflow_invocation: + # Running workflow invocation in background, just mark + # persistent workflow invocation as failed. + workflow_invocation.fail() + trans.sa_session.add( workflow_invocation ) + else: + # Running new transient workflow invocation in legacy + # controller action - propage the exception up. + raise + outputs = [] + + if workflow_invocation: + # Be sure to update state of workflow_invocation. + trans.sa_session.flush() + + return outputs, invoker.workflow_invocation + + +def queue_invoke( trans, workflow, workflow_run_config, request_params ): + modules.populate_module_and_state( trans, workflow, workflow_run_config.param_map ) + workflow_invocation = workflow_run_config_to_request( trans, workflow_run_config, workflow ) + workflow_invocation.workflow = workflow + return trans.app.workflow_scheduling_manager.queue( + workflow_invocation, + request_params + ) class WorkflowInvoker( object ): - def __init__( self, trans, workflow, workflow_run_config ): + def __init__( self, trans, workflow, workflow_run_config, workflow_invocation=None ): self.trans = trans self.workflow = workflow - workflow_invocation = model.WorkflowInvocation() - workflow_invocation.workflow = self.workflow - self.workflow_invocation = workflow_invocation - self.progress = WorkflowProgress( self.workflow_invocation, workflow_run_config.inputs ) + if workflow_invocation is None: + invocation_uuid = uuid.uuid1() - invocation_uuid = uuid.uuid1().hex + workflow_invocation = model.WorkflowInvocation() + workflow_invocation.workflow = self.workflow + + # In one way or another, following attributes will become persistent + # so they are available during delayed/revisited workflow scheduling. + workflow_invocation.uuid = invocation_uuid + workflow_invocation.history = workflow_run_config.target_history + + self.workflow_invocation = workflow_invocation + else: + self.workflow_invocation = workflow_invocation - # In one way or another, following attributes will become persistent - # so they are available during delayed/revisited workflow scheduling. - self.workflow_invocation.uuid = invocation_uuid - self.workflow_invocation.history = workflow_run_config.target_history self.workflow_invocation.copy_inputs_to_history = workflow_run_config.copy_inputs_to_history self.workflow_invocation.replacement_dict = workflow_run_config.replacement_dict + module_injector = modules.WorkflowModuleInjector( trans ) + self.progress = WorkflowProgress( self.workflow_invocation, workflow_run_config.inputs, module_injector ) + def invoke( self ): workflow_invocation = self.workflow_invocation remaining_steps = self.progress.remaining_steps() + delayed_steps = False for step in remaining_steps: - jobs = self._invoke_step( step ) - for job in util.listify( jobs ): - # Record invocation - workflow_invocation_step = model.WorkflowInvocationStep() - workflow_invocation_step.workflow_invocation = workflow_invocation - workflow_invocation_step.workflow_step = step - workflow_invocation_step.job = job + jobs = None + try: + jobs = self._invoke_step( step ) + for job in (util.listify( jobs ) or [None]): + # Record invocation + workflow_invocation_step = model.WorkflowInvocationStep() + workflow_invocation_step.workflow_invocation = workflow_invocation + workflow_invocation_step.workflow_step = step + workflow_invocation_step.job = job + except modules.DelayedWorkflowEvaluation: + delayed_steps = True + self.progress.mark_step_outputs_delayed( step ) + + if delayed_steps: + state = model.WorkflowInvocation.states.READY + else: + state = model.WorkflowInvocation.states.SCHEDULED + workflow_invocation.state = state # All jobs ran successfully, so we can save now self.trans.sa_session.add( workflow_invocation ) @@ -66,18 +122,35 @@ class WorkflowInvoker( object ): jobs = step.module.execute( self.trans, self.progress, self.workflow_invocation, step ) return jobs +STEP_OUTPUT_DELAYED = object() + class WorkflowProgress( object ): - def __init__( self, workflow_invocation, inputs_by_step_id ): + def __init__( self, workflow_invocation, inputs_by_step_id, module_injector ): self.outputs = odict() + self.module_injector = module_injector self.workflow_invocation = workflow_invocation self.inputs_by_step_id = inputs_by_step_id 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 + remaining_steps = [] + step_invocations_by_id = self.workflow_invocation.step_invocations_by_step_id() + for step in steps: + if not hasattr( step, 'module' ): + self.module_injector.inject( step ) + runtime_state = step_states[ step.id ].value + step.state = step.module.recover_runtime_state( runtime_state ) - return steps + invocation_steps = step_invocations_by_id.get( step.id, None ) + if invocation_steps: + self._recover_mapping( step, invocation_steps ) + else: + remaining_steps.append( step ) + return remaining_steps def replacement_for_tool_input( self, step, input, prefixed_name ): """ For given workflow 'step' that has had input_connections_by_name @@ -100,6 +173,8 @@ class WorkflowProgress( object ): def replacement_for_connection( self, connection ): step_outputs = self.outputs[ connection.output_step.id ] + if step_outputs is STEP_OUTPUT_DELAYED: + raise modules.DelayedWorkflowEvaluation() return step_outputs[ connection.output_name ] def set_outputs_for_input( self, step, outputs={} ): @@ -111,5 +186,13 @@ class WorkflowProgress( object ): def set_step_outputs(self, step, outputs): self.outputs[ step.id ] = outputs + def mark_step_outputs_delayed(self, step): + self.outputs[ step.id ] = STEP_OUTPUT_DELAYED + + def _recover_mapping( self, step, step_invocations ): + try: + step.module.recover_mapping( step, step_invocations, self ) + except modules.DelayedWorkflowEvaluation: + self.mark_step_outputs_delayed( step ) __all__ = [ invoke, WorkflowRunConfig ] diff --git a/lib/galaxy/workflow/run_request.py b/lib/galaxy/workflow/run_request.py index 3a9a475eba2..88b9bd01e57 100644 --- a/lib/galaxy/workflow/run_request.py +++ b/lib/galaxy/workflow/run_request.py @@ -1,9 +1,15 @@ +import uuid + from galaxy import exceptions +from galaxy import model from galaxy.managers import histories INPUT_STEP_TYPES = [ 'data_input', 'data_collection_input' ] +import logging +log = logging.getLogger( __name__ ) + class WorkflowRunConfig( object ): """ Wrapper around all the ways a workflow execution can be parameterized. @@ -231,6 +237,96 @@ def build_workflow_run_config( trans, workflow, payload ): return run_config +def workflow_run_config_to_request( trans, run_config, workflow ): + param_types = model.WorkflowRequestInputParameter.types + + workflow_invocation = model.WorkflowInvocation() + workflow_invocation.uuid = uuid.uuid1() + workflow_invocation.history = run_config.target_history + + def add_parameter( name, value, type ): + parameter = model.WorkflowRequestInputParameter( + name=name, + value=value, + type=type, + ) + workflow_invocation.input_parameters.append( parameter ) + + replacement_dict = run_config.replacement_dict + for name, value in replacement_dict.iteritems(): + add_parameter( + name=name, + value=value, + type=param_types.REPLACEMENT_PARAMETERS, + ) + + for step_id, content in run_config.inputs.iteritems(): + if content.history_content_type == "dataset": + request_to_content = model.WorkflowRequestToInputDatasetAssociation() + request_to_content.dataset = content + request_to_content.workflow_step_id = step_id + workflow_invocation.input_datasets.append( request_to_content ) + else: + request_to_content = model.WorkflowRequestToInputDatasetCollectionAssociation() + request_to_content.dataset_collection = content + request_to_content.workflow_step_id = step_id + workflow_invocation.input_dataset_collections.append( request_to_content ) + + for step in workflow.steps: + state = step.state + serializable_runtime_state = step.module.normalize_runtime_state( state ) + step_state = model.WorkflowRequestStepState() + step_state.workflow_step_id = step.id + step_state.value = serializable_runtime_state + workflow_invocation.step_states.append( step_state ) + + add_parameter( "copy_inputs_to_history", "true" if run_config.copy_inputs_to_history else "false", param_types.META_PARAMETERS ) + return workflow_invocation + + +def workflow_request_to_run_config( work_request_context, workflow_invocation ): + param_types = model.WorkflowRequestInputParameter.types + + history = workflow_invocation.history + replacement_dict = {} + inputs = {} + param_map = {} + copy_inputs_to_history = None + + for parameter in workflow_invocation.input_parameters: + parameter_type = parameter.type + + if parameter_type == param_types.REPLACEMENT_PARAMETERS: + replacement_dict[ parameter.name ] = parameter.value + elif parameter_type == param_types.META_PARAMETERS: + if parameter.name == "copy_inputs_to_history": + copy_inputs_to_history = (parameter.value == "true") + + #for parameter in workflow_invocation.step_parameters: + # step_id = parameter.workflow_step_id + # if step_id not in param_map: + # param_map[ step_id ] = {} + # param_map[ step_id ][ parameter.name ] = parameter.value + + for input_association in workflow_invocation.input_datasets: + inputs[ input_association.workflow_step_id ] = input_association.dataset + + for input_association in workflow_invocation.input_dataset_collections: + inputs[ input_association.workflow_step_id ] = input_association.dataset_collection + + if copy_inputs_to_history is None: + raise exceptions.InconsistentDatabase("Failed to find copy_inputs_to_history parameter loading workflow_invocation from database.") + + workflow_run_config = WorkflowRunConfig( + target_history=history, + replacement_dict=replacement_dict, + inputs=inputs, + param_map=param_map, + copy_inputs_to_history=copy_inputs_to_history, + ) + return workflow_run_config + + def __decode_id( trans, workflow_id, model_type="workflow" ): try: return trans.security.decode_id( workflow_id ) diff --git a/lib/galaxy/workflow/schedulers/__init__.py b/lib/galaxy/workflow/schedulers/__init__.py new file mode 100644 index 00000000000..a7836e16198 --- /dev/null +++ b/lib/galaxy/workflow/schedulers/__init__.py @@ -0,0 +1,41 @@ +""" Module containing Galaxy workflow scheduling plugins. Galaxy's interface +for workflow scheduling is highly experimental and the interface required for +scheduling plugins will almost certainly change. +""" +from abc import ABCMeta +from abc import abstractmethod + + +class WorkflowSchedulingPlugin( object ): + """ A plugin defining how Galaxy should schedule plugins. By default + plugins are passive and should monitor Galaxy's work queue for + WorkflowRequests. Inherit from ActiveWorkflowSchedulingPlugin instead if + the scheduling plugin should be forced (i.e. if scheduling happen all at + once or the request will be stored and monitored outside of Galaxy.) + """ + __metaclass__ = ABCMeta + + @property + @abstractmethod + def plugin_type( self ): + """ Short string providing labelling this plugin """ + + def startup( self, app ): + """ Called when Galaxy starts up if the plugin is enabled. + """ + + def shutdown( self ): + """ Called when Galaxy is shutting down, workflow scheduling should + end. + """ + + +class ActiveWorkflowSchedulingPlugin( WorkflowSchedulingPlugin ): + __metaclass__ = ABCMeta + + @abstractmethod + def schedule( self, workflow_invocation ): + """ Optionally return one or more commands to instrument job. These + commands will be executed on the compute server prior to the job + running. + """ diff --git a/lib/galaxy/workflow/schedulers/core.py b/lib/galaxy/workflow/schedulers/core.py new file mode 100644 index 00000000000..4a5fbd24e6a --- /dev/null +++ b/lib/galaxy/workflow/schedulers/core.py @@ -0,0 +1,46 @@ +""" The class defines the stock Galaxy workflow scheduling plugin - currently +it simply schedules the whole workflow up front when offered. +""" +from ..schedulers import ActiveWorkflowSchedulingPlugin + +from galaxy.work import context + +from galaxy.workflow import run +from galaxy.workflow import run_request + +import logging +log = logging.getLogger( __name__ ) + + +class CoreWorkflowSchedulingPlugin( ActiveWorkflowSchedulingPlugin ): + plugin_type = "core" + + def __init__( self, **kwds ): + pass + + def startup( self, app ): + self.app = app + + def shutdown( self ): + pass + + def schedule( self, workflow_invocation ): + workflow = workflow_invocation.workflow + history = workflow_invocation.history + request_context = context.WorkRequestContext( + app=self.app, + history=history, + user=history.user + ) # trans-like object not tied to a web-thread. + workflow_run_config = run_request.workflow_request_to_run_config( + request_context, + workflow_invocation + ) + run.invoke( + trans=request_context, + workflow=workflow, + workflow_run_config=workflow_run_config, + workflow_invocation=workflow_invocation, + ) + +__all__ = [ CoreWorkflowSchedulingPlugin ] diff --git a/lib/galaxy/workflow/scheduling_manager.py b/lib/galaxy/workflow/scheduling_manager.py new file mode 100644 index 00000000000..28bfaaf08f6 --- /dev/null +++ b/lib/galaxy/workflow/scheduling_manager.py @@ -0,0 +1,197 @@ +import os +import time +import logging +import threading + +from xml.etree import ElementTree + +from galaxy import model +from galaxy.util import plugin_config + +import galaxy.workflow.schedulers + +log = logging.getLogger( __name__ ) + +DEFAULT_SCHEDULER_ID = "default" # well actually this should be called DEFAULT_DEFAULT_SCHEDULER_ID... +DEFAULT_SCHEDULER_PLUGIN_TYPE = "core" + +EXCEPTION_MESSAGE_SHUTDOWN = "Exception raised while attempting to shutdown workflow scheduler." +EXCEPTION_MESSAGE_NO_SCHEDULERS = "Failed to defined workflow schedulers - no workflow schedulers defined." +EXCEPTION_MESSAGE_NO_DEFAULT_SCHEDULER = "Failed to defined workflow schedulers - no workflow scheduler found for default id '%s'." +EXCEPTION_MESSAGE_DUPLICATE_SCHEDULERS = "Failed to defined workflow schedulers - workflow scheduling plugin id '%s' duplicated." + + +class WorkflowSchedulingManager( object ): + """ A workflow scheduling manager based loosely on pattern established by + ``galaxy.manager.JobManager``. Only schedules workflows on handler + processes. + """ + + def __init__( self, app ): + self.app = app + self.__job_config = app.job_config + self.workflow_schedulers = {} + self.active_workflow_schedulers = {} # Passive workflow schedulers + # won't need to be monitored I + # guess. + self.request_monitor = None + + self.__plugin_classes = self.__plugins_dict() + self.__init_schedulers() + + if self._is_workflow_handler(): + log.debug("Starting workflow schedulers") + self.__start_schedulers() + if self.active_workflow_schedulers: + self.__start_request_monitor() + else: + # Process should not schedule workflows - do nothing. + pass + + # Provide a handler config-like interface by delegating to job handler + # config. Perhaps it makes sense to let there be explicit workflow + # handlers? + def _is_workflow_handler( self ): + return self.app.is_job_handler() + + def _get_handler( self ): + return self.__job_config.get_handler( None ) + + def shutdown( self ): + for workflow_scheduler in self.workflow_schedulers.itervalues(): + try: + workflow_scheduler.shutdown() + except Exception: + log.exception( EXCEPTION_MESSAGE_SHUTDOWN ) + if self.request_monitor: + try: + self.request_monitor.shutdown() + except Exception: + log.exception( "Failed to shutdown worklfow request monitor." ) + + def queue( self, workflow_invocation, request_params ): + workflow_invocation.state = model.WorkflowInvocation.states.NEW + scheduler = request_params.get( "scheduler", None ) or self.default_scheduler_id + handler = self._get_handler() + + workflow_invocation.scheduler = scheduler + workflow_invocation.handler = handler + + sa_session = self.app.model.context + sa_session.add( workflow_invocation ) + sa_session.flush() + return workflow_invocation + + def __start_schedulers( self ): + for workflow_scheduler in self.workflow_schedulers.itervalues(): + workflow_scheduler.startup( self.app ) + + def __plugins_dict( self ): + return plugin_config.plugins_dict( galaxy.workflow.schedulers, 'plugin_type' ) + + def __init_schedulers( self ): + config_file = self.app.config.workflow_schedulers_config_file + use_default_scheduler = False + if not config_file: + log.info( "Not workflow schedulers plugin config file defined, using default scheduler." ) + use_default_scheduler = True + elif not os.path.exists( config_file ): + log.info( "Cannot find workflow schedulers plugin config file '%s', using default scheduler." % config_file ) + use_default_scheduler = True + + if use_default_scheduler: + self.__init_default_scheduler() + else: + plugins_element = ElementTree.parse( config_file ).getroot() + self.__init_schedulers_for_element( plugins_element ) + + def __init_default_scheduler( self ): + self.default_scheduler_id = DEFAULT_SCHEDULER_ID + self.__init_plugin( DEFAULT_SCHEDULER_PLUGIN_TYPE ) + + def __init_schedulers_for_element( self, plugins_element ): + plugins_kwds = dict( plugins_element.items() ) + self.default_scheduler_id = plugins_kwds.get( 'default', DEFAULT_SCHEDULER_ID ) + + for plugin_element in plugins_element.getchildren(): + plugin_type = plugin_element.tag + plugin_kwds = dict( plugin_element.items() ) + plugin_kwds.update( self.extra_kwargs ) + workflow_scheduler_id = plugin_kwds.get( 'id', None ) + self.__init_plugin( plugin_type, workflow_scheduler_id, **plugin_kwds ) + + if not self.workflow_schedulers: + raise Exception( EXCEPTION_MESSAGE_NO_SCHEDULERS ) + if self.default_scheduler_id not in self.workflow_schedulers: + raise Exception( EXCEPTION_MESSAGE_NO_DEFAULT_SCHEDULER % self.default_scheduler_id ) + + def __init_plugin( self, plugin_type, workflow_scheduler_id=None, **kwds ): + workflow_scheduler_id = workflow_scheduler_id or self.default_scheduler_id + + if workflow_scheduler_id in self.workflow_schedulers: + raise Exception( EXCEPTION_MESSAGE_DUPLICATE_SCHEDULERS % workflow_scheduler_id ) + + workflow_scheduler = self.__plugin_classes[ plugin_type ]( **kwds ) + self.workflow_schedulers[ workflow_scheduler_id ] = workflow_scheduler + if isinstance( workflow_scheduler, galaxy.workflow.schedulers.ActiveWorkflowSchedulingPlugin ): + self.active_workflow_schedulers[ workflow_scheduler_id ] = workflow_scheduler + + def __start_request_monitor( self ): + self.request_monitor = WorkflowRequestMonitor( self.app, self ) + + +class WorkflowRequestMonitor( object ): + + def __init__( self, app, workflow_scheduling_manager ): + self.app = app + self.active = True + self.workflow_scheduling_manager = workflow_scheduling_manager + self.monitor_thread = threading.Thread( name="WorkflowRequestMonitor.monitor_thread", target=self.__monitor ) + self.monitor_thread.setDaemon( True ) + self.monitor_thread.start() + + def __monitor( self ): + to_monitor = self.workflow_scheduling_manager.active_workflow_schedulers + while self.active: + for workflow_scheduler_id, workflow_scheduler in to_monitor.iteritems(): + if not self.active: + return + + self.__schedule( workflow_scheduler_id, workflow_scheduler ) + time.sleep(1) # TODO: wake if stopped + + def __schedule( self, workflow_scheduler_id, workflow_scheduler ): + invocation_ids = self.__active_invocation_ids( workflow_scheduler_id ) + for invocation_id in invocation_ids: + self.__attempt_schedule( invocation_id, workflow_scheduler ) + if not self.active: + return + + def __attempt_schedule( self, invocation_id, workflow_scheduler ): + sa_session = self.app.model.context + workflow_invocation = sa_session.query( model.WorkflowInvocation ).get( invocation_id ) + + if not workflow_invocation or not workflow_invocation.active: + return False + + try: + workflow_scheduler.schedule( workflow_invocation ) + except Exception: + # TODO: eventually fail this - or fail it right away? + log.exception( "Exception raised while attempting to schedule workflow request." ) + return False + + # A workflow was obtained and scheduled... + return True + + def __active_invocation_ids( self, scheduler_id ): + sa_session = self.app.model.context + handler = self.app.config.server_name + return model.WorkflowInvocation.poll_active_workflow_ids( + sa_session, + scheduler=scheduler_id, + handler=handler, + ) + + def shutdown( self ): + self.active = False diff --git a/test/api/test_workflows.py b/test/api/test_workflows.py index dab0146425d..c064377cb8a 100644 --- a/test/api/test_workflows.py +++ b/test/api/test_workflows.py @@ -135,9 +135,28 @@ class WorkflowsApiTestCase( api.ApiTestCase ): # TODO: This should really be a post to workflows//run or # something like that. run_workflow_response = self._post( "workflows", data=workflow_request ) + + invocation_id = run_workflow_response.json()[ "id" ] + invocation = self._invocation_details( workflow_request[ "workflow_id" ], invocation_id ) + assert invocation[ "state" ] == "scheduled", invocation + self._assert_status_code_is( run_workflow_response, 200 ) self.dataset_populator.wait_for_history( history_id, assert_ok=True ) + def test_workflow_request( self ): + workflow = self.workflow_populator.load_workflow( name="test_for_queue" ) + workflow_request, history_id = self._setup_workflow_run( workflow ) + # TODO: This should really be a post to workflows//run or + # something like that. + url = "workflows/%s/request" % ( workflow_request[ "workflow_id" ] ) + del workflow_request[ "workflow_id" ] + run_workflow_response = self._post( url, data=workflow_request ) + + self._assert_status_code_is( run_workflow_response, 200 ) + # Give some time for workflow to get scheduled before scanning the history. + time.sleep( 5 ) + self.dataset_populator.wait_for_history( history_id, assert_ok=True ) + def test_cannot_run_inaccessible_workflow( self ): workflow = self.workflow_populator.load_workflow( name="test_for_run_cannot_access" ) workflow_request, history_id = self._setup_workflow_run( workflow ) @@ -563,11 +582,18 @@ class WorkflowsApiTestCase( api.ApiTestCase ): @skip_without_tool( "cat1" ) def test_invocation_usage( self ): workflow_id, usage = self._run_workflow_once_get_invocation( "test_usage") - usage_details = self._invocation_details( workflow_id, usage[ "id" ] ) + invocation_id = usage[ "id" ] + usage_details = self._invocation_details( workflow_id, invocation_id ) # Assert some high-level things about the structure of data returned. self._assert_has_keys( usage_details, "inputs", "steps" ) - for step in usage_details[ "steps" ]: + invocation_steps = usage_details[ "steps" ] + for step in invocation_steps: self._assert_has_keys( step, "workflow_step_id", "order_index", "id" ) + an_invocation_step = invocation_steps[ 0 ] + step_id = an_invocation_step[ "id" ] + step_response = self._get( "workflows/%s/usage/%s/steps/%s" % ( workflow_id, invocation_id, step_id ) ) + self._assert_status_code_is( step_response, 200 ) + self._assert_has_keys( step_response.json(), "id", "order_index" ) def _invocation_details( self, workflow_id, invocation_id ): invocation_details_response = self._get( "workflows/%s/usage/%s" % ( workflow_id, invocation_id ) ) diff --git a/test/unit/test_galaxy_mapping.py b/test/unit/test_galaxy_mapping.py index 4aa1672bbce..468b0a3d91b 100644 --- a/test/unit/test_galaxy_mapping.py +++ b/test/unit/test_galaxy_mapping.py @@ -1,6 +1,7 @@ # -*- coding: utf-8 -*- import unittest import galaxy.model.mapping as mapping +import uuid class MappingTests( unittest.TestCase ): @@ -358,6 +359,54 @@ class MappingTests( unittest.TestCase ): assert contents_iter_names( ids=[ d1.id, d3.id ] ) == [ "1", "3" ] + def test_workflows( self ): + model = self.model + user = model.User( + email="testworkflows@bx.psu.edu", + password="password" + ) + stored_workflow = model.StoredWorkflow() + stored_workflow.user = user + workflow = model.Workflow() + workflow_step = model.WorkflowStep() + workflow.steps = [ workflow_step ] + workflow.stored_workflow = stored_workflow + + self.persist( workflow ) + assert workflow_step.id is not None + + invocation_uuid = uuid.uuid1() + + workflow_invocation = model.WorkflowInvocation() + workflow_invocation.uuid = invocation_uuid + + workflow_invocation_step1 = model.WorkflowInvocationStep() + workflow_invocation_step1.workflow_invocation = workflow_invocation + workflow_invocation_step1.workflow_step = workflow_step + + workflow_invocation_step2 = model.WorkflowInvocationStep() + workflow_invocation_step2.workflow_invocation = workflow_invocation + workflow_invocation_step2.workflow_step = workflow_step + + workflow_invocation.workflow = workflow + + h1 = model.History( name="WorkflowHistory1", user=user) + d1 = self.new_hda( h1, name="1" ) + workflow_request_dataset = model.WorkflowRequestToInputDatasetAssociation() + workflow_request_dataset.workflow_invocation = workflow_invocation + workflow_request_dataset.workflow_step = workflow_step + workflow_request_dataset.dataset = d1 + self.persist( workflow_invocation ) + assert workflow_request_dataset is not None + assert workflow_invocation.id is not None + + self.expunge() + + loaded_invocation = self.query( model.WorkflowInvocation ).get( workflow_invocation.id ) + assert loaded_invocation.uuid == invocation_uuid, "%s != %s" % (loaded_invocation.uuid, invocation_uuid) + assert loaded_invocation + assert len( loaded_invocation.steps ) == 2 + def new_hda( self, history, **kwds ): return history.add_dataset( self.model.HistoryDatasetAssociation( create_dataset=True, sa_session=self.model.session, **kwds ) ) diff --git a/test/unit/workflows/__init__.py b/test/unit/workflows/__init__.py new file mode 100644 index 00000000000..e69de29bb2d