diff --git a/job_conf.xml.sample_advanced b/job_conf.xml.sample_advanced index 9e0b589b627..8d149fb0b88 100644 --- a/job_conf.xml.sample_advanced +++ b/job_conf.xml.sample_advanced @@ -62,6 +62,11 @@ + + diff --git a/job_metrics_conf.xml.sample b/job_metrics_conf.xml.sample new file mode 100644 index 00000000000..66774e9780b --- /dev/null +++ b/job_metrics_conf.xml.sample @@ -0,0 +1,18 @@ + + + + + + + + diff --git a/lib/galaxy/app.py b/lib/galaxy/app.py index 13c90fa1a27..574fdf38b02 100644 --- a/lib/galaxy/app.py +++ b/lib/galaxy/app.py @@ -15,7 +15,7 @@ from galaxy.tools.genome_index import load_genome_index_tools from galaxy.sample_tracking import external_service_types from galaxy.openid.providers import OpenIDProviders from galaxy.tools.data_manager.manager import DataManagers - +from galaxy.jobs import metrics as job_metrics from galaxy.web.base import pluginframework import logging @@ -60,6 +60,10 @@ class UniverseApplication( object, config.ConfiguresGalaxyMixin ): self._configure_tool_data_tables( from_shed_config=False ) + # Initialize job metrics manager, needs to be in place before + # config so per-destination modifications can be made. + self.job_metrics = job_metrics.JobMetrics( self.config.job_metrics_config_file, app=self ) + # Initialize the job management configuration self.job_config = jobs.JobConfiguration(self) diff --git a/lib/galaxy/config.py b/lib/galaxy/config.py index 72767feabf2..b47b2a30de2 100644 --- a/lib/galaxy/config.py +++ b/lib/galaxy/config.py @@ -137,6 +137,7 @@ class Configuration( object ): 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.dependency_resolvers_config_file = resolve_path( kwargs.get( 'dependency_resolvers_config_file', 'dependency_resolvers_conf.xml' ), self.root ) + self.job_metrics_config_file = resolve_path( kwargs.get( 'job_metrics_config_file', 'job_metrics_conf.xml' ), self.root ) self.job_config_file = resolve_path( kwargs.get( 'job_config_file', 'job_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" ) ) @@ -453,6 +454,11 @@ class Configuration( object ): admin_users = [ x.strip() for x in self.get( "admin_users", "" ).split( "," ) ] return ( user is not None and user.email in admin_users ) + def resolve_path( self, path ): + """ Resolve a path relative to Galaxy's root. + """ + return resolve_path( path, self.root ) + def get_database_engine_options( kwargs, model_prefix='' ): """ Allow options for the SQLAlchemy database engine to be passed by using diff --git a/lib/galaxy/jobs/__init__.py b/lib/galaxy/jobs/__init__.py index 190cbc42882..47ac2f70bee 100644 --- a/lib/galaxy/jobs/__init__.py +++ b/lib/galaxy/jobs/__init__.py @@ -162,8 +162,21 @@ class JobConfiguration( object ): # Parse destinations destinations = root.find('destinations') + job_metrics = self.app.job_metrics for destination in self.__findall_with_required(destinations, 'destination', ('id', 'runner')): id = destination.get('id') + destination_metrics = destination.get( "metrics", None ) + if destination_metrics: + if not util.asbool( destination_metrics ): + # disable + job_metrics.set_destination_instrumenter( id, None ) + else: + metrics_conf_path = self.app.config.resolve_path( destination_metrics ) + job_metrics.set_destination_conf_file( id, metrics_conf_path ) + else: + metrics_elements = self.__findall_with_required( destination, 'job_metrics', () ) + if metrics_elements: + job_metrics.set_destination_conf_element( id, metrics_elements[ 0 ] ) job_destination = JobDestination(**dict(destination.items())) job_destination['params'] = self.__get_params(destination) self.destinations[id] = (job_destination,) @@ -1068,8 +1081,10 @@ class JobWrapper( object ): # Finally set the job state. This should only happen *after* all # dataset creation, and will allow us to eliminate force_history_refresh. job.state = final_job_state + if not job.tasks: + # If job was composed of tasks, don't attempt to recollect statisitcs + self._collect_metrics( job ) self.sa_session.flush() - log.debug( 'job %d ended' % self.job_id ) delete_files = self.app.config.cleanup_job == 'always' or ( job.state == job.states.OK and self.app.config.cleanup_job == 'onsuccess' ) self.cleanup( delete_files=delete_files ) @@ -1094,6 +1109,16 @@ class JobWrapper( object ): except: log.exception( "Unable to cleanup job %d" % self.job_id ) + def _collect_metrics( self, has_metrics ): + job = has_metrics.get_job() + per_plugin_properties = self.app.job_metrics.collect_properties( job.destination_id, self.job_id, self.working_directory ) + if per_plugin_properties: + log.info( "Collecting job metrics for %s" % has_metrics ) + for plugin, properties in per_plugin_properties.iteritems(): + for metric_name, metric_value in properties.iteritems(): + if metric_value is not None: + has_metrics.add_metric( plugin, metric_name, metric_value ) + def get_output_sizes( self ): sizes = [] output_paths = self.get_output_fnames() @@ -1508,6 +1533,7 @@ class TaskWrapper(JobWrapper): task.stdout = util.shrink_string_by_size( stdout, DATABASE_MAX_STRING_SIZE, join_by="\n..\n", left_larger=True, beginning_on_size_error=True ) if len( stderr ) > DATABASE_MAX_STRING_SIZE: log.error( "stderr for task %d is greater than %s, only a portion will be logged to database" % ( task.id, DATABASE_MAX_STRING_SIZE_PRETTY ) ) + self._collect_metrics( task ) task.stderr = util.shrink_string_by_size( stderr, DATABASE_MAX_STRING_SIZE, join_by="\n..\n", left_larger=True, beginning_on_size_error=True ) task.exit_code = tool_exit_code task.command_line = self.command_line diff --git a/lib/galaxy/jobs/metrics/__init__.py b/lib/galaxy/jobs/metrics/__init__.py new file mode 100644 index 00000000000..19001622adb --- /dev/null +++ b/lib/galaxy/jobs/metrics/__init__.py @@ -0,0 +1,134 @@ +import collections +import os + +from xml.etree import ElementTree + +from galaxy.util.submodules import submodules +from galaxy import util + +from ..metrics import formatting + +import logging +log = logging.getLogger( __name__ ) + + +DEFAULT_FORMATTER = formatting.JobMetricFormatter() + + +class JobMetrics( object ): + + def __init__( self, conf_file=None, **kwargs ): + """ + """ + self.plugin_classes = self.__plugins_dict() + self.default_job_instrumenter = JobInstrumenter.from_file( self.plugin_classes, conf_file, **kwargs ) + self.job_instrumenters = collections.defaultdict( lambda: self.default_job_instrumenter ) + + def format( self, plugin, key, value ): + if plugin in self.plugin_classes: + plugin_class = self.plugin_classes[ plugin ] + formatter = plugin_class.formatter + else: + formatter = DEFAULT_FORMATTER + return formatter.format( key, value ) + + def set_destination_conf_file( self, destination_id, conf_file ): + instrumenter = JobInstrumenter.from_file( self.plugin_classes, conf_file ) + self.set_destination_instrumenter( destination_id, instrumenter ) + + def set_destination_conf_element( self, destination_id, element ): + instrumenter = JobInstrumenter( self.plugin_classes, element ) + self.set_destination_instrumenter( destination_id, instrumenter ) + + def set_destination_instrumenter( self, destination_id, job_instrumenter=None ): + if job_instrumenter is None: + job_instrumenter = NULL_JOB_INSTRUMENTER + self.job_instrumenters[ destination_id ] = job_instrumenter + + def collect_properties( self, destination_id, job_id, job_directory ): + return self.job_instrumenters[ destination_id ].collect_properties( job_id, job_directory ) + + def __plugins_dict( self ): + plugin_dict = {} + for plugin_module in self.__plugin_modules(): + for clazz in plugin_module.__all__: + plugin_type = getattr( clazz, 'plugin_type', None ) + if plugin_type: + plugin_dict[ plugin_type ] = clazz + return plugin_dict + + def __plugin_modules( self ): + import galaxy.jobs.metrics.instrumenters + return submodules( galaxy.jobs.metrics.instrumenters ) + + +class NullJobInstrumenter( object ): + + def pre_execute_commands( self, job_directory ): + return None + + def post_execute_commands( self, job_directory ): + return None + + def collect_properties( self, job_id, job_directory ): + return {} + +NULL_JOB_INSTRUMENTER = NullJobInstrumenter() + + +class JobInstrumenter( object ): + + def __init__( self, plugin_classes, metrics_element, **kwargs ): + self.extra_kwargs = kwargs + self.plugin_classes = plugin_classes + self.plugins = self.__plugins_for_element( metrics_element ) + + def pre_execute_commands( self, job_directory ): + commands = [] + for plugin in self.plugins: + try: + plugin_commands = plugin.pre_execute_instrument( job_directory ) + if plugin_commands: + commands.extend( util.listify( plugin_commands ) ) + except Exception: + log.exception( "Failed to generate pre-execute commands for plugin %s" % plugin ) + return "\n".join( [ c for c in commands if c ] ) + + def post_execute_commands( self, job_directory ): + commands = [] + for plugin in self.plugins: + try: + plugin_commands = plugin.post_execute_instrument( job_directory ) + if plugin_commands: + commands.extend( util.listify( plugin_commands ) ) + except Exception: + log.exception( "Failed to generate post-execute commands for plugin %s" % plugin ) + return "\n".join( [ c for c in commands if c ] ) + + def collect_properties( self, job_id, job_directory ): + per_plugin_properites = {} + for plugin in self.plugins: + try: + properties = plugin.job_properties( job_id, job_directory ) + if properties: + per_plugin_properites[ plugin.plugin_type ] = properties + except Exception: + log.exception( "Failed to collect job properties for plugin %s" % plugin ) + return per_plugin_properites + + def __plugins_for_element( self, plugins_element ): + plugins = [] + for plugin_element in plugins_element.getchildren(): + plugin_type = plugin_element.tag + plugin_kwds = dict( plugin_element.items() ) + plugin_kwds.update( self.extra_kwargs ) + plugin = self.plugin_classes[ plugin_type ]( **plugin_kwds ) + plugins.append( plugin ) + return plugins + + @staticmethod + def from_file( plugin_classes, conf_file, **kwargs ): + if not conf_file or not os.path.exists( conf_file ): + return NULL_JOB_INSTRUMENTER + plugins_element = ElementTree.parse( conf_file ).getroot() + return JobInstrumenter( plugin_classes, plugins_element, **kwargs ) diff --git a/lib/galaxy/jobs/metrics/formatting.py b/lib/galaxy/jobs/metrics/formatting.py new file mode 100644 index 00000000000..b3679c3650b --- /dev/null +++ b/lib/galaxy/jobs/metrics/formatting.py @@ -0,0 +1,18 @@ + + +class JobMetricFormatter( object ): + """ Format job metric key-value pairs for human consumption in Web UI. """ + + def format( self, key, value ): + return ( str( key ), str( value ) ) + + +## Formatting utilities + +def seconds_to_str( value ): + if value < 60: + return "%s seconds" % value + elif value < 3600: + return "%s minutes" % ( value / 60 ) + else: + return "%s days and %s minutes" % ( value / 3600, ( value % 3600 ) / 60 ) diff --git a/lib/galaxy/jobs/metrics/instrumenters/__init__.py b/lib/galaxy/jobs/metrics/instrumenters/__init__.py new file mode 100644 index 00000000000..c461cdb1970 --- /dev/null +++ b/lib/galaxy/jobs/metrics/instrumenters/__init__.py @@ -0,0 +1,53 @@ +from abc import ABCMeta +from abc import abstractmethod + +import os.path + +from ...metrics import formatting + + +INSTRUMENT_FILE_PREFIX = "__instrument" + + +class InstrumentPlugin( object ): + """ A plugin describing how to instrument Galaxy jobs and retrieve metrics + from this instrumentation. + """ + __metaclass__ = ABCMeta + formatter = formatting.JobMetricFormatter() + + @property + @abstractmethod + def plugin_type( self ): + """ Short string providing labelling this plugin """ + + def pre_execute_instrument( self, job_directory ): + """ Optionally return one or more commands to instrument job. These + commands will be executed on the compute server prior to the job + running. + """ + return None + + def post_execute_instrument( self, job_directory ): + """ Optionally return one or more commands to instrument job. These + commands will be executed on the compute server after the tool defined + command is ran. + """ + return None + + @abstractmethod + def job_properties( self, job_id, job_directory ): + """ Collect properties for this plugin from specified job directory. + This method will run on the Galaxy server and can assume files created + in job_directory with pre_execute_instrument and + post_execute_instrument are available. + """ + + def _instrument_file_name( self, name ): + """ Provide a common pattern for naming files used by instrumentation + plugins - to ease their staging out of remote job directories. + """ + return "%s_%s_%s" % ( INSTRUMENT_FILE_PREFIX, self.plugin_type, name ) + + def _instrument_file_path( self, job_directory, name ): + return os.path.join( job_directory, self._instrument_file_name( name ) ) diff --git a/lib/galaxy/jobs/metrics/instrumenters/core.py b/lib/galaxy/jobs/metrics/instrumenters/core.py new file mode 100644 index 00000000000..2b6e2021c07 --- /dev/null +++ b/lib/galaxy/jobs/metrics/instrumenters/core.py @@ -0,0 +1,85 @@ +from ..instrumenters import InstrumentPlugin +from ...metrics import formatting +import time + +import logging +log = logging.getLogger( __name__ ) + +GALAXY_SLOTS_KEY = "galaxy_slots" +START_EPOCH_KEY = "start_epoch" +END_EPOCH_KEY = "end_epoch" +RUNTIME_SECONDS_KEY = "runtime_seconds" + + +class CorePluginFormatter( formatting.JobMetricFormatter ): + + def format( self, key, value ): + value = int( value ) + if key == GALAXY_SLOTS_KEY: + return ( "Cores Allocated", "%d" % value ) + elif key == RUNTIME_SECONDS_KEY: + return ( "Job Runtime (Wall Clock)", formatting.seconds_to_str( value ) ) + else: + # TODO: Use localized version of this from universe_wsgi.ini + title = "Job Start Time" if key == START_EPOCH_KEY else "Job End Time" + return (title, time.strftime( '%Y-%m-%d %H:%M:%S', time.localtime( value ) ) ) + + +class CorePlugin( InstrumentPlugin ): + """ Simple plugin that collects data without external dependencies. In + particular it currently collects value set for Galaxy slots. + """ + plugin_type = "core" + formatter = CorePluginFormatter() + + def __init__( self, **kwargs ): + pass + + def pre_execute_instrument( self, job_directory ): + commands = [] + commands.append( self.__record_galaxy_slots_command( job_directory ) ) + commands.append( self.__record_seconds_since_epoch_to_file( job_directory, "start" ) ) + return commands + + def post_execute_instrument( self, job_directory ): + commands = [] + commands.append( self.__record_seconds_since_epoch_to_file( job_directory, "end" ) ) + return commands + + def job_properties( self, job_id, job_directory ): + galaxy_slots_file = self.__galaxy_slots_file( job_directory ) + + properties = {} + properties[ GALAXY_SLOTS_KEY ] = self.__read_integer( galaxy_slots_file ) + start = self.__read_seconds_since_epoch( job_directory, "start" ) + end = self.__read_seconds_since_epoch( job_directory, "end" ) + if start is not None and end is not None: + properties[ START_EPOCH_KEY ] = start + properties[ END_EPOCH_KEY ] = end + properties[ RUNTIME_SECONDS_KEY ] = end - start + return properties + + def __record_galaxy_slots_command( self, job_directory ): + galaxy_slots_file = self.__galaxy_slots_file( job_directory ) + return '''echo "$GALAXY_SLOTS" > '%s' ''' % galaxy_slots_file + + def __record_seconds_since_epoch_to_file( self, job_directory, name ): + path = self._instrument_file_path( job_directory, "epoch_%s" % name ) + return 'date +"%s" > ' + path + + def __read_seconds_since_epoch( self, job_directory, name ): + path = self._instrument_file_path( job_directory, "epoch_%s" % name ) + return self.__read_integer( path ) + + def __galaxy_slots_file( self, job_directory ): + return self._instrument_file_path( job_directory, "galaxy_slots" ) + + def __read_integer( self, path ): + value = None + try: + value = int( open( path, "r" ).read() ) + except Exception: + pass + return value + +__all__ = [ CorePlugin ] diff --git a/lib/galaxy/jobs/runners/__init__.py b/lib/galaxy/jobs/runners/__init__.py index 76b5c63327a..efc376d26c4 100644 --- a/lib/galaxy/jobs/runners/__init__.py +++ b/lib/galaxy/jobs/runners/__init__.py @@ -251,7 +251,10 @@ class BaseJobRunner( object ): log.debug( 'execution of external set_meta for job %d finished' % job_wrapper.job_id ) def get_job_file(self, job_wrapper, **kwds): + job_metrics = job_wrapper.app.job_metrics + job_instrumenter = job_metrics.job_instrumenters[ job_wrapper.job_destination.id ] options = dict( + job_instrumenter=job_instrumenter, galaxy_lib=job_wrapper.galaxy_lib_dir, env_setup_commands=job_wrapper.get_env_setup_clause(), working_directory=os.path.abspath( job_wrapper.working_directory ), diff --git a/lib/galaxy/jobs/runners/util/job_script/DEFAULT_JOB_FILE_TEMPLATE.sh b/lib/galaxy/jobs/runners/util/job_script/DEFAULT_JOB_FILE_TEMPLATE.sh index 26006c7f640..28af9118358 100644 --- a/lib/galaxy/jobs/runners/util/job_script/DEFAULT_JOB_FILE_TEMPLATE.sh +++ b/lib/galaxy/jobs/runners/util/job_script/DEFAULT_JOB_FILE_TEMPLATE.sh @@ -12,6 +12,8 @@ if [ "$GALAXY_LIB" != "None" ]; then export PYTHONPATH fi $env_setup_commands +$instrument_pre_commands cd $working_directory $command echo $? > $exit_code_path +$instrument_post_commands diff --git a/lib/galaxy/jobs/runners/util/job_script/__init__.py b/lib/galaxy/jobs/runners/util/job_script/__init__.py index a60ec19ae08..55c9b5fcf3e 100644 --- a/lib/galaxy/jobs/runners/util/job_script/__init__.py +++ b/lib/galaxy/jobs/runners/util/job_script/__init__.py @@ -18,6 +18,8 @@ OPTIONAL_TEMPLATE_PARAMS = { 'headers': '', 'env_setup_commands': '', 'slots_statement': SLOTS_STATEMENT_CLUSTER_DEFAULT, + 'instrument_pre_commands': '', + 'instrument_post_commands': '', } @@ -47,6 +49,13 @@ def job_script(template=DEFAULT_JOB_FILE_TEMPLATE, **kwds): """ if any([param not in kwds for param in REQUIRED_TEMPLATE_PARAMS]): raise Exception("Failed to create job_script, a required parameter is missing.") + job_instrumenter = kwds.get("job_instrumenter", None) + if job_instrumenter: + del kwds[ "job_instrumenter" ] + working_directory = kwds["working_directory"] + kwds["instrument_pre_commands"] = job_instrumenter.pre_execute_commands(working_directory) or '' + kwds["instrument_post_commands"] = job_instrumenter.post_execute_commands(working_directory) or '' + template_params = OPTIONAL_TEMPLATE_PARAMS.copy() template_params.update(**kwds) if not isinstance(template, Template): diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py index 6aa712794e8..aa54b13ffda 100644 --- a/lib/galaxy/model/__init__.py +++ b/lib/galaxy/model/__init__.py @@ -102,6 +102,11 @@ class HasJobMetrics: metric = self._text_metric( plugin, metric_name, metric_value ) self.text_metrics.append( metric ) + @property + def metrics( self ): + # TODO: Make iterable, concatenate with chain + return self.text_metrics + self.numeric_metrics + class User( object, Dictifiable ): use_pbkdf2 = True @@ -414,6 +419,12 @@ class Job( object, HasJobMetrics, Dictifiable ): self.info = info def set_runner_name( self, job_runner_name ): self.job_runner_name = job_runner_name + + def get_job( self ): + # Added so job and task have same interface (.get_job() ) to get at + # underlying job object. + return self + def set_runner_external_id( self, job_runner_external_id ): self.job_runner_external_id = job_runner_external_id def set_post_job_actions( self, post_job_actions ): diff --git a/templates/show_params.mako b/templates/show_params.mako index 094ae37730a..b4952f64360 100644 --- a/templates/show_params.mako +++ b/templates/show_params.mako @@ -126,6 +126,13 @@ %if job and job.command_line and trans.user_is_admin(): Job Command-Line:${ job.command_line | h } %endif + %if job and trans.user_is_admin(): + <% job_metrics = trans.app.job_metrics %> + %for metric in job.metrics: + <% metric_title, metric_value = job_metrics.format( metric.plugin, metric.metric_name, metric.metric_value ) %> + ${ metric_title | h }${ metric_value | h } + %endfor + %endif
diff --git a/test/unit/test_galaxy_mapping.py b/test/unit/test_galaxy_mapping.py index 9311286af14..1622fe7b99b 100644 --- a/test/unit/test_galaxy_mapping.py +++ b/test/unit/test_galaxy_mapping.py @@ -234,6 +234,11 @@ class MappingTests( unittest.TestCase ): self.persist( u, job ) + task = model.Task( job=job, working_directory="/tmp", prepare_files_cmd="split.sh" ) + task.add_metric( "gx", "galaxy_slots", 5 ) + task.add_metric( "system", "system_name", "localhost" ) + self.persist( task ) + def test_tasks( self ): model = self.model u = model.User( email="jobtest@foo.bar.baz", password="password" )