Implement plugin framework for collecting data about runtime job execution.

An example job_metrics_conf.xml.sample is included that describes which plugins are enabled and how they are configured. This will be updated for each new plugin added. By default not instrumentation or data collection occurs - but if a job_metrics.xml file is present it will serve as the default for all job destination. Additionally, individual job destinations may disable, load a different job metrics file, or define metrics directly in job_conf.xml in an embedded fashion. See comment at top of job_metrics_conf.xml for more information.

This commit include an initial plugin (named 'core') to demonstrate the framework and capture the highest priority data - namely the number of cores allocated to the job and the runtime of the job on the cluster. These two pieces of information alone should provide a much clearer picture of what Galaxy is actually allocating cluster compute cycles to.

Current limitations - This only works with job runners utilizing the job script module and the LWR (it utilizes the job script module on the remote server), hence it won't yet work with...

 - Local job runner - I do have a downstream fork of Galaxy where I have reworked the local job runner to use the common job script template.

    https://github.com/jmchilton/galaxy-central/commits/local_job_script
    https://github.com/jmchilton/galaxy-central/commit/949db2cd14c7191cedf1febeb5f4a2f53123123e

 - CLI runner - CLI runner needs to be reworked to use the job script module anyway so GALAXY_SLOTS works - the LWR version of the CLI runner uses the job script module - this work just needs to be back ported to Galaxy.

If a job_metrics_conf.xml is present and some jobs route to the above destinations - the jobs won't fail but annoying errors will appear in the logs. Simply attach a 'metrics="off"' those these specific job destinations to disable any attempt to use metrics for these jobs and disable these errors.

job_conf.xml.sample_advanced

@@ -62,6 +62,11 @@
           <!-- Warning: Local slot count doesn't tie up additional worker threads, to prevent over 
                allocating machine define a second local runner with different name and fewer workers
                to run this destination. -->
+          <job_metrics /> 
+          <!-- Above element demonstrates embedded job metrics definition - see
+               job_metrics_conf.xml.sample for full documentation on possible nested
+               elements. This empty block will simply disable job metrics for the
+               corresponding destination. -->
         </destination>
         <destination id="pbs" runner="pbs" tags="mycluster"/>
         <destination id="pbs_longjobs" runner="pbs" tags="mycluster,longjobs">

job_metrics_conf.xml.sample

@@ -0,0 +1,18 @@
+<?xml version="1.0"?>
+<!-- If job_metrics.xml exists, this file will define the default job metric
+     plugin used for all jobs. Individual job_conf.xml destinations can
+     disable metric collection by setting metrics="off" on that destination.
+     The metrics attribute on destination definition elements can also be
+     a path - in which case that XML metrics file will be loaded and used for
+     that destination. Finally, the destination element may contain a job_metrics
+     child element (with all options defined below) to define job metrics in an
+     embedded manner directly in the job_conf.xml file.
+-->
+<job_metrics>
+  <!-- Each element in this file corresponds to a job instrumentation plugin
+       used to generate metrics in lib/galaxy/jobs/metrics/instrumenters. -->
+
+  <!-- Core plugin captures Galaxy slots, start and end of job (in seconds
+       since epoch) and computes runtime in seconds. -->
+  <core />
+</job_metrics>

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)
 

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

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

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 )

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 )

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

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 ]

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

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

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

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

templates/show_params.mako

@@ -126,6 +126,13 @@
         %if job and job.command_line and trans.user_is_admin():
             <tr><td>Job Command-Line:</td><td>${ job.command_line | h }</td></tr>
         %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 ) %>
+                <tr><td>${ metric_title | h }</td><td>${ metric_value | h }</td></tr>
+            %endfor
+        %endif
 </table>
 <br />
 

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