<commit_before>"""
Library to generate statistics from the new Stampede 3.1 backend.

Usage::
 stats = StampedeStatistics(connString='sqlite:///montage.db')
 stats.initialize('unique_wf_uuid')
 stats.set_job_filter('dax')
 print stats.get_total_jobs_status()
 print stats.get_total_succeeded_jobs_status()
 stats.set_job_filter('dag')
 print stats.get_total_jobs_status()
 print stats.get_total_succeeded_jobs_status()
 etc.
 stats.close()
 
Constructor and initialize methods:

The constructor takes a required sqlalchemy connection string
as the first argument.  The stats class will default to returning
data in the "expanded workflow" mode.  To change this behavior
and only analyize a single workflow set the optional arg:

expand_workflow = False

along with the connection string argument.

The initialize method is called with a single argument - the wf_uuid
of the desired "root workflow" whether returning data in expanded
mode or not.  The method will return True or False if a query
exception is raised so the programmer can test for success before
calling the subsequent query methods.  This method is intended
to be called once per object.

Job filtering:

Jobs can be filtered using any of the strings in the jobtype ENUM, 
with the addition of the values 'all' and 'nonsub' which will
return all jobs and non-subworkflow jobs respectively.  If the 
filter is not explicitly set, it will default to the 'all' mode.

The desired filter can be set with the set_job_filter() method. After
setting this method, all subsequent calls to the query methods will
return results according to the filter.  This can be set and reset
as many times as the user desires.  There is an example of re/setting 
the job filter in the usage section above.  The query methods
will return different values after the filter is re/set.

Time filtering:

This behaves much like job filtering.  For the runtime queries, 
the time intervals 'month', 'week', 'day', and  'hour' can 
be set using the set_time_filter() method.  If this method
is not set, it will default to the 'month' interval for filtering.

Hostname filtering:

For the runtime queries the method set_host_filter() can be used to 
filter by various hosts.  This method differs from the job and time
filtering methods in that the argument can be either a string (for
a single hostname), or an array/list of hostnames for multiple 
hostnames.

Example::
 s.set_host_filter('butterfly.isi.edu')
 or
 s.set_host_filter(['engage-submit3.renci.org', 'node0012.palmetto.clemson.edu'])
 
Either one of these variations will work.  The first variation will
only retrieve data for that one host, the second will return data
for both hosts.  If this method is not set, no hostname filtering 
will be done and information for all hosts will be returned.

Transformation filtering:

Transformation filtering works similarly to hostname filtering in 
that it can accept a single string value or a array/list of strings. 
However the set_transformation_filter() method accepts two keyword
arguments - 'include' and 'exclude'.  Only one of these keywords can 
be set per method call.

Example::
 s.set_transformation_filter(include='pegasus::dirmanager')
 s.set_transformation_filter(exclude=['dagman::post' , 'dagman::pre' ,'condor::dagman'])
 etc.

This example demonstrates the two proper keyword invocations and 
that either a string or list may be used.  If this method is not 
set, no filtering will be done and information for all transforms 
will be returned.  Calling this method with no arguments will  
reset any previously set filters.

Return values from methods:

The return value types will vary from method to method.  Most of
the methods will return a single integer or floating point number.

Methods which return rows from the DB (rather than just a number) 
will return a list which can be interacted with in one of two 
ways - either by array index (list of tuples) or by a named attr
(list of objects).  The two following methods of interacting with 
the same query results will both produce the same output:

Example::
 for row in s.get_job_kickstart():
     print row[0], row[1], row[2]
     print row.job_id, row.job_name, row.kickstart

Either syntax will work.  When using the named attribute method, the
attributes are the names of the columns/aliases in the SELECT 
stanza of the query.  If the row returned by the method is printed, 
it will display as a tuple of results per row.

Methods::
 get_sub_workflow_ids
 get_descendant_workflow_ids
 get_schema_version
 get_total_jobs_status
 get_total_succeeded_jobs_status
 get_total_failed_jobs_status
 get_total_jobs_retries
 get_total_tasks_status
 get_total_succeeded_tasks_status
 get_total_failed_tasks_status
 get_task_success_report
 get_task_failure_report
 get_total_tasks_retries
 get_workflow_states
 get_workflow_cum_job_wall_time
 get_submit_side_job_wall_time
 get_workflow_details
 get_workflow_retries
 get_job_statistics
 get_job_states
 get_job_instance_sub_wf_map
 get_failed_job_instances
 get_job_instance_info
 get_job_name
 get_job_site
 get_job_kickstart
 get_job_runtime
 get_job_seqexec
 get_condor_q_time
 get_resource_delay
 get_post_time
 get_transformation_statistics
 get_invocation_by_time
 get_jobs_run_by_time
 get_invocation_by_time_per_host
 get_jobs_run_by_time_per_host
 
Methods listed in order of query list on wiki.

https://confluence.pegasus.isi.edu/display/pegasus/Pegasus+Statistics+Python+Version+Modified
"""
__rcsid__ = "$Id: stampede_statistics.py 31115 2012-03-27 19:35:06Z mgoode $"
__author__ = "<NAME>"

from netlogger.analysis.modules._base import SQLAlchemyInit
from netlogger.analysis.schema.schema_check import ErrorStrings, SchemaCheck, SchemaVersionError
from netlogger.analysis.schema.stampede_schema import *
from netlogger.nllog import DoesLogging, get_logger
        
# Main stats class.

class StampedeStatistics(SQLAlchemyInit, DoesLogging):
    def __init__(self, connString=None, expand_workflow=True):
        if connString is None:
            raise ValueError("connString is required")
        DoesLogging.__init__(self)
        try:
            SQLAlchemyInit.__init__(self, connString, initializeToPegasusDB)
        except exceptions.OperationalError, e:
            self.log.error('init', msg='%s' % ErrorStrings.get_init_error(e))
            raise RuntimeError
            
        # Check the schema version before proceeding.
        self.s_check = SchemaCheck(self.session)
        if not self.s_check.check_schema():
            raise SchemaVersionError
        
        self._expand = expand_workflow
        
        self._root_wf_id = None
        self._root_wf_uuid = None
        self._job_filter_mode = None
        self._time_filter_mode = None
        self._host_filter = None
        self._xform_filter = {'include':None, 'exclude':None}
        
        self._wfs = []
        pass
    
    def initialize(self, root_wf_uuid):
        self.log.debug('initialize')
        self._root_wf_uuid = root_wf_uuid
        q = self.session.query(Workflow.wf_id).filter(Workflow.wf_uuid == self._root_wf_uuid)      
        
        try:
            self._root_wf_id = q.one().wf_id
        except orm.exc.MultipleResultsFound, e:
            self.log.error('initialize',
                msg='Multiple results found for wf_uuid: %s' % root_wf_uuid)
            return False
        except orm.exc.NoResultFound, e:
            self.log.error('initialize',
                msg='No results found for wf_uuid: %s' % root_wf_uuid)
            return False
        
        if self._expand:
            q = self.session.query(Workflow.wf_id).filter(Workflow.root_wf_id == self._root_wf_id)
            for row in q.all():
                self._wfs.append(row.wf_id)
            if not len(self._wfs):
                self.log.error('initialize',
                    msg='Unable to expand wf_uuid: %s - not a root_wf_id?' % root_wf_uuid)
                return False
        else:
            self._wfs.append(self._root_wf_id)
            
        if not len(self._wfs):
            self.log.error('initialize',
                msg='No results found for wf_uuid: %s' % root_wf_uuid)
            return False
        
        # Initialize filters with default value
        self.set_job_filter()
        self.set_time_filter()
        self.set_host_filter()
        self.set_transformation_filter()
        return True
        
    def close(self):
        self.log.debug('close')
        self.disconnect()
        
    def set_job_filter(self, filter='all'):
        modes = ['all', 'nonsub', 'subwf', 'dax', 'dag', 'compute', 'stage-in-tx', 
                'stage-out-tx', 'registration', 'inter-site-tx', 'create-dir', 
                'staged-compute', 'cleanup', 'chmod']
        try:
            modes.index(filter)
            self._job_filter_mode = filter
            self.log.debug('set_job_filter', msg='Setting filter to: %s' % filter)
        except:
            self._job_filter_mode = 'all'
            self.log.error('set_job_filter', msg='Unknown job filter %s - setting to all' % filter)
            
            
    def set_time_filter(self, filter='month'):
        modes = ['month', 'week', 'day', 'hour']
        try:
            modes.index(filter)
            self._time_filter_mode = filter
            self.log.debug('set_time_filter', msg='Setting filter to: %s' % filter)
        except:
            self._job_filter_mode = 'month'
            self.log.error('set_time_filter', msg='Unknown time filter %s - setting to month' % filter)
            
    def set_host_filter(self, host=None):
        """
        The host argument can either be a string/single hostname or
        it can be a list/array of hostnames.
        """
        self._host_filter = host
        
    def set_transformation_filter(self, include=None, exclude=None):
        """
        Either of these args can either be a single string/xform type or
        it can be a list/array of xform types.
        
        Both arguments can not be set at the same time.  If they are,
        the program will log an error and not do any filtering.
        """
        self._xform_filter['include'] = include
        self._xform_filter['exclude'] = exclude
                
    #
    # Pulls information about sub workflows
    #
            
    def get_sub_workflow_ids(self):
        """
        Returns info on child workflows only.
        """
        q = self.session.query(Workflow.wf_id, Workflow.wf_uuid, Workflow.dax_label)
        q = q.filter(Workflow.parent_wf_id == self._root_wf_id)
        return q.all()
        
    def get_descendant_workflow_ids(self):
        q = self.session.query(Workflow.wf_id, Workflow.wf_uuid)
        q = q.filter(Workflow.root_wf_id == self._root_wf_id)
        q = q.filter(Workflow.wf_id != self._root_wf_id)
        return q.all()
        
    def get_schema_version(self):
        return self.s_check.check_version()
            
    #
    # Status of initially planned wf components.
    #
    
    #
    # The following block of queries are documented here:
    # https://confluence.pegasus.isi.edu/display/pegasus/Workflow+Summary
    # and
    # https://confluence.pegasus.isi.edu/display/pegasus/Workflow+Statistics+file
    #
    
    def _dax_or_dag_cond(self, JobO=Job):
        return or_(JobO.type_desc == 'dax', JobO.type_desc == 'dag')
        
    def _get_job_filter(self, JobO=Job):
        filters = {
            'all': None,
            'nonsub': not_(self._dax_or_dag_cond(JobO)),
            'subwf': self._dax_or_dag_cond(JobO),
            'dax': JobO.type_desc == 'dax',
            'dag': JobO.type_desc == 'dag',
            'compute': JobO.type_desc == 'compute',
            'stage-in-tx': JobO.type_desc == 'stage-in-tx',
            'stage-out-tx': JobO.type_desc == 'stage-out-tx',
            'registration': JobO.type_desc == 'registration',
            'inter-site-tx': JobO.type_desc == 'inter-site-tx',
            'create-dir': JobO.type_desc == 'create-dir',
            'staged-compute': JobO.type_desc == 'staged-compute',
            'cleanup': JobO.type_desc == 'cleanup',
            'chmod': JobO.type_desc == 'chmod',
        }
        return filters[self._job_filter_mode]
        
    def _max_job_seq_subquery(self):
        """
        Creates the following subquery that is used in 
        several queries:
        and jb_inst.job_submit_seq  = (
            select max(job_submit_seq) from job_instance where job_id = jb_inst.job_id group by job_id
            )
        """
        JobInstanceSubMax = orm.aliased(JobInstance)
        sub_q = self.session.query(func.max(JobInstanceSubMax.job_submit_seq).label('max_id'))
        sub_q = sub_q.filter(JobInstanceSubMax.job_id == JobInstance.job_id).correlate(JobInstance)
        sub_q = sub_q.group_by(JobInstanceSubMax.job_id).subquery()
        return sub_q
        
    def get_total_jobs_status(self):
        """
        https://confluence.pegasus.isi.edu/display/pegasus/Workflow+Summary#WorkflowSummary-Totaljobs
        https://confluence.pegasus.isi.edu/display/pegasus/Workflow+Statistics+file#WorkflowStatisticsfile-Totaljobs        
        """
        q = self.session.query(Job.job_id)
        if self._expand:
            q = q.filter(Workflow.root_wf_id == self._root_wf_id)
        else:
            q = q.filter(Workflow.wf_id == self._wfs[0])
        q = q.filter(Job.wf_id == Workflow.wf_id)
        if self._get_job_filter() is not None:
            q = q.filter(self._get_job_filter())
             
        return q.count()
    
        
    def get_total_succeeded_jobs_status(self):
        """
        https://confluence.pegasus.isi.edu/display/pegasus/Workflow+Summary#WorkflowSummary-Totalsucceededjobs
        https://confluence.pegasus.isi.edu/display/pegasus/Workflow+Statistics+file#WorkflowStatisticsfile-Totalsucceededjobs
        """
        JobInstanceSub = orm.aliased(JobInstance, name='JobInstanceSub')
        sq_1 = self.session.query(func.max(JobInstanceSub.job_submit_seq).label('jss'), JobInstanceSub.job_id.label('jobid'),
                JobInstanceSub.exitcode.label('ec'))
        if self._expand:
            sq_1 = sq_1.filter(Workflow.root_wf_id == self._root_wf_id)
        else:
            sq_1 = sq_1.filter(Workflow.wf_id == self._wfs[0])
        sq_1 = sq_1.filter(Workflow.wf_id == Job.wf_id)
        sq_1 = sq_1.filter(Job.job_id == JobInstanceSub.job_id)
        if self._get_job_filter() is not None:
            sq_1 = sq_1.filter(self._get_job_filter())
        sq_1 = sq_1.group_by(JobInstanceSub.job_id).subquery()
        
        q = self.session.query(JobInstance.job_instance_id.label('last_job_instance'))
        q = q.filter(JobInstance.job_id == sq_1.c.jobid)
        q = q.filter(JobInstance.job_submit_seq == sq_1.c.jss)
        q = q.filter(sq_1.c.ec == 0).filter(sq_1.c.ec != None)
        
        return q.count()
    
        
    def get_total_failed_jobs_status(self):
        """
        https://confluence.pegasus.isi.edu/display/pegasus/Workflow+Summary#WorkflowSummary-Totalfailedjobs
        https://confluence.pegasus.isi.edu/display/pegasus/Workflow+Statistics+file#WorkflowStatisticsfile-Totalfailedjobs
        """
        JobInstanceSub = orm.aliased(JobInstance, name='JobInstanceSub')
        sq_1 = self.session.query(func.max(JobInstanceSub.job_submit_seq).label('jss'), JobInstanceSub.job_id.label('jobid'),
                JobInstanceSub.exitcode.label('ec'))
        if self._expand:
            sq_1 = sq_1.filter(Workflow.root_wf_id == self._root_wf_id)
        else:
            sq_1 = sq_1.filter(Workflow.wf_id == self._wfs[0])
        sq_1 = sq_1.filter(Workflow.wf_id == Job.wf_id)
        sq_1 = sq_1.filter(Job.job_id == JobInstanceSub.job_id)
        if self._get_job_filter() is not None:
            sq_1 = sq_1.filter(self._get_job_filter())
        sq_1 = sq_1.group_by(JobInstanceSub.job_id).subquery()
        
        q = self.session.query(JobInstance.job_instance_id.label('last_job_instance'))
        q = q.filter(JobInstance.job_id == sq_1.c.jobid)
        q = q.filter(JobInstance.job_submit_seq == sq_1.c.jss)
        q = q.filter(sq_1.c.ec != 0).filter(sq_1.c.ec != None)
        
        return q.count()
        
    def _query_jobstate_for_instance(self, states):
        """
        The states arg is a list of strings.
        Returns an appropriate subquery.
        """
        q = self.session.query(Jobstate.job_instance_id)
        q = q.filter(Jobstate.job_instance_id == JobInstance.job_instance_id).correlate(JobInstance)
        q = q.filter(Jobstate.state.in_(states)).subquery()
        return q
        
    def get_total_jobs_retries(self):
        """
        https://confluence.pegasus.isi.edu/display/pegasus/Workflow+Summary#WorkflowSummary-TotalJobRetries
        https://confluence.pegasus.isi.edu/display/pegasus/Workflow+Statistics+file#WorkflowStatisticsfile-TotalJobRetries
        """
        d_or_d = self._dax_or_dag_cond()
        
        sq_1 = self.session.query(func.count(Job.job_id))
        if self._expand:
            sq_1 = sq_1.filter(Workflow.root_wf_id == self._root_wf_id)
        else:
            sq_1 = sq_1.filter(Workflow.wf_id == self._wfs[0])
        sq_1 = sq_1.filter(Job.wf_id == Workflow.wf_id)
        sq_1 = sq_1.filter(Job.job_id == JobInstance.job_id)
        if self._get_job_filter() is not None:
            sq_1 = sq_1.filter(self._get_job_filter())

        sq_1 = sq_1.subquery()

        
        sq_2 = self.session.query(func.count(distinct(JobInstance.job_id)))
        if self._expand:
            sq_2 = sq_2.filter(Workflow.root_wf_id == self._root_wf_id)
        else:
            sq_2 = sq_2.filter(Workflow.wf_id == self._wfs[0])
        sq_2 = sq_2.filter(Job.wf_id == Workflow.wf_id)
        sq_2 = sq_2.filter(Job.job_id == JobInstance.job_id)
        if self._get_job_filter() is not None:
            sq_2 = sq_2.filter(self._get_job_filter())

        sq_2 = sq_2.subquery()
        
        q = self.session.query((sq_1.as_scalar() - sq_2.as_scalar()).label('total_job_retries'))
        
        return q.all()[0].total_job_retries
        
    def get_total_tasks_status(self):
        """
        https://confluence.pegasus.isi.edu/display/pegasus/Workflow+Summary#WorkflowSummary-Totaltask
        https://confluence.pegasus.isi.edu/display/pegasus/Workflow+Statistics+file#WorkflowStatisticsfile-Totaltasks
        """
        q = self.session.query(Task.task_id)
        if self._expand:
            q = q.filter(Workflow.root_wf_id == self._root_wf_id)
        else:
            q = q.filter(Workflow.wf_id == self._wfs[0])
        q = q.filter(Task.wf_id == Workflow.wf_id)
        q = q.filter(Task.job_id == Job.job_id)
        if self._get_job_filter(Task) is not None:
            q = q.filter(self._get_job_filter(Task))
        return q.count()
    
        
    def _base_task_status_query_old(self):
        """
        https://confluence.pegasus.isi.edu/display/pegasus/Workflow+Summary#WorkflowSummary-Totalsucceededtasks
        https://confluence.pegasus.isi.edu/display/pegasus/Workflow+Statistics+file#WorkflowStatisticsfile-Totalsucceededtasks
        """
        # This query generation method is obsolete and is only being
        # kept for optimization reference.
        
        WorkflowSub1 = orm.aliased(Workflow, name='WorkflowSub1')
        JobInstanceSub1 = orm.aliased(JobInstance, name='JobInstanceSub1')
        JobSub1 = orm.aliased(Job, name='JobSub1')
        
        sq_1 = self.session.query(WorkflowSub1.wf_id.label('wid'), 
                func.max(JobInstanceSub1.job_submit_seq).label('jss'),
                JobInstanceSub1.job_id.label('jobid')
        )
        if self._expand:
            sq_1 = sq_1.filter(WorkflowSub1.root_wf_id == self._root_wf_id)
        else:
            sq_1 = sq_1.filter(WorkflowSub1.wf_id == self._wfs[0])
        sq_1 = sq_1.filter(WorkflowSub1.wf_id == JobSub1.wf_id)
        sq_1 = sq_1.filter(JobSub1.job_id == JobInstanceSub1.job_id)
        sq_1 = sq_1.group_by(JobInstanceSub1.job_id)
        if self._get_job_filter(JobSub1) is not None:
            sq_1 = sq_1.filter(self._get_job_filter(JobSub1))
        sq_1 = sq_1.subquery()
        
        JobInstanceSub2 = orm.aliased(JobInstance, name='JobInstanceSub2')
        sq_2 = self.session.query(sq_1.c.wid.label('wf_id'), JobInstanceSub2.job_instance_id.label('last_job_instance_id'))
        sq_2 = sq_2.filter(JobInstanceSub2.job_id == sq_1.c.jobid)
        sq_2 = sq_2.filter(JobInstanceSub2.job_submit_seq == sq_1.c.jss)
        sq_2 = sq_2.subquery()
        
        q = self.session.query(Invocation.invocation_id)
        q = q.filter(Invocation.abs_task_id != None)
        q = q.filter(Invocation.job_instance_id == sq_2.c.last_job_instance_id)
        q = q.filter(Invocation.wf_id == sq_2.c.wf_id)
        
        # Calling wrapper methods would invoke like so:
        # q = self._base_task_status_query()
        # q = q.filter(Invocation.exitcode == 0)
        # return q.count()
        
        return q
        
    def _base_task_statistics_query(self, success=True):
        w = orm.aliased(Workflow, name='w')
        j = orm.aliased(Job, name='j')
        ji = orm.aliased(JobInstance, name='ji')

        sq_1 = self.session.query(w.wf_id, 
                j.job_id,
                ji.job_instance_id.label('jiid'),
                ji.job_submit_seq.label('jss'),
                func.max(ji.job_submit_seq).label('maxjss'))
        sq_1 = sq_1.join(j, w.wf_id == j.wf_id)
        sq_1 = sq_1.join(ji, j.job_id == ji.job_id)
        if self._expand:
            sq_1 = sq_1.filter(w.root_wf_id == self._root_wf_id)
        else:
            sq_1 = sq_1.filter(w.wf_id == self._wfs[0])
        sq_1 = sq_1.group_by(j.job_id)
        sq_1 = sq_1.subquery('t')

        sq_2 = self.session.query(sq_1.c.wf_id, func.count(Invocation.exitcode).label('count'))
        sq_2 = sq_2.select_from(orm.join(sq_1, Invocation, sq_1.c.jiid == Invocation.job_instance_id))
        sq_2 = sq_2.filter(sq_1.c.jss == sq_1.c.maxjss)
        sq_2 = sq_2.filter(Invocation.abs_task_id != None)
        if success:
            sq_2 = sq_2.filter(Invocation.exitcode == 0)
        else:
            sq_2 = sq_2.filter(Invocation.exitcode != 0)
        sq_2 = sq_2.group_by(sq_1.c.wf_id)
        
        return sq_2
        
    def _task_statistics_query_sum(self, success=True):
        s = self._base_task_statistics_query(success).subquery('tt')
        q = self.session.query(func.sum(s.c.count).label('task_count'))
        return q.one()[0] or 0
    
    def get_total_succeeded_tasks_status(self):
        return self._task_statistics_query_sum(success=True)

    def get_total_failed_tasks_status(self):<commit_msg>Fixes the negative task count during sub workflows<commit_after>"""
Library to generate statistics from the new Stampede 3.1 backend.

Usage::
 stats = StampedeStatistics(connString='sqlite:///montage.db')
 stats.initialize('unique_wf_uuid')
 stats.set_job_filter('dax')
 print stats.get_total_jobs_status()
 print stats.get_total_succeeded_jobs_status()
 stats.set_job_filter('dag')
 print stats.get_total_jobs_status()
 print stats.get_total_succeeded_jobs_status()
 etc.
 stats.close()
 
Constructor and initialize methods:

The constructor takes a required sqlalchemy connection string
as the first argument.  The stats class will default to returning
data in the "expanded workflow" mode.  To change this behavior
and only analyize a single workflow set the optional arg:

expand_workflow = False

along with the connection string argument.

The initialize method is called with a single argument - the wf_uuid
of the desired "root workflow" whether returning data in expanded
mode or not.  The method will return True or False if a query
exception is raised so the programmer can test for success before
calling the subsequent query methods.  This method is intended
to be called once per object.

Job filtering:

Jobs can be filtered using any of the strings in the jobtype ENUM, 
with the addition of the values 'all' and 'nonsub' which will
return all jobs and non-subworkflow jobs respectively.  If the 
filter is not explicitly set, it will default to the 'all' mode.

The desired filter can be set with the set_job_filter() method. After
setting this method, all subsequent calls to the query methods will
return results according to the filter.  This can be set and reset
as many times as the user desires.  There is an example of re/setting 
the job filter in the usage section above.  The query methods
will return different values after the filter is re/set.

Time filtering:

This behaves much like job filtering.  For the runtime queries, 
the time intervals 'month', 'week', 'day', and  'hour' can 
be set using the set_time_filter() method.  If this method
is not set, it will default to the 'month' interval for filtering.

Hostname filtering:

For the runtime queries the method set_host_filter() can be used to 
filter by various hosts.  This method differs from the job and time
filtering methods in that the argument can be either a string (for
a single hostname), or an array/list of hostnames for multiple 
hostnames.

Example::
 s.set_host_filter('butterfly.isi.edu')
 or
 s.set_host_filter(['engage-submit3.renci.org', 'node0012.palmetto.clemson.edu'])
 
Either one of these variations will work.  The first variation will
only retrieve data for that one host, the second will return data
for both hosts.  If this method is not set, no hostname filtering 
will be done and information for all hosts will be returned.

Transformation filtering:

Transformation filtering works similarly to hostname filtering in 
that it can accept a single string value or a array/list of strings. 
However the set_transformation_filter() method accepts two keyword
arguments - 'include' and 'exclude'.  Only one of these keywords can 
be set per method call.

Example::
 s.set_transformation_filter(include='pegasus::dirmanager')
 s.set_transformation_filter(exclude=['dagman::post' , 'dagman::pre' ,'condor::dagman'])
 etc.

This example demonstrates the two proper keyword invocations and 
that either a string or list may be used.  If this method is not 
set, no filtering will be done and information for all transforms 
will be returned.  Calling this method with no arguments will  
reset any previously set filters.

Return values from methods:

The return value types will vary from method to method.  Most of
the methods will return a single integer or floating point number.

Methods which return rows from the DB (rather than just a number) 
will return a list which can be interacted with in one of two 
ways - either by array index (list of tuples) or by a named attr
(list of objects).  The two following methods of interacting with 
the same query results will both produce the same output:

Example::
 for row in s.get_job_kickstart():
     print row[0], row[1], row[2]
     print row.job_id, row.job_name, row.kickstart

Either syntax will work.  When using the named attribute method, the
attributes are the names of the columns/aliases in the SELECT 
stanza of the query.  If the row returned by the method is printed, 
it will display as a tuple of results per row.

Methods::
 get_sub_workflow_ids
 get_descendant_workflow_ids
 get_schema_version
 get_total_jobs_status
 get_total_succeeded_jobs_status
 get_total_failed_jobs_status
 get_total_jobs_retries
 get_total_tasks_status
 get_total_succeeded_tasks_status
 get_total_failed_tasks_status
 get_task_success_report
 get_task_failure_report
 get_total_tasks_retries
 get_workflow_states
 get_workflow_cum_job_wall_time
 get_submit_side_job_wall_time
 get_workflow_details
 get_workflow_retries
 get_job_statistics
 get_job_states
 get_job_instance_sub_wf_map
 get_failed_job_instances
 get_job_instance_info
 get_job_name
 get_job_site
 get_job_kickstart
 get_job_runtime
 get_job_seqexec
 get_condor_q_time
 get_resource_delay
 get_post_time
 get_transformation_statistics
 get_invocation_by_time
 get_jobs_run_by_time
 get_invocation_by_time_per_host
 get_jobs_run_by_time_per_host
 
Methods listed in order of query list on wiki.

https://confluence.pegasus.isi.edu/display/pegasus/Pegasus+Statistics+Python+Version+Modified
"""
__rcsid__ = "$Id: stampede_statistics.py 31185 2012-04-16 17:57:46Z mgoode $"
__author__ = "<NAME>"

from netlogger.analysis.modules._base import SQLAlchemyInit
from netlogger.analysis.schema.schema_check import ErrorStrings, SchemaCheck, SchemaVersionError
from netlogger.analysis.schema.stampede_schema import *
from netlogger.nllog import DoesLogging, get_logger
        
# Main stats class.

class StampedeStatistics(SQLAlchemyInit, DoesLogging):
    def __init__(self, connString=None, expand_workflow=True):
        if connString is None:
            raise ValueError("connString is required")
        DoesLogging.__init__(self)
        try:
            SQLAlchemyInit.__init__(self, connString, initializeToPegasusDB)
        except exceptions.OperationalError, e:
            self.log.error('init', msg='%s' % ErrorStrings.get_init_error(e))
            raise RuntimeError
            
        # Check the schema version before proceeding.
        self.s_check = SchemaCheck(self.session)
        if not self.s_check.check_schema():
            raise SchemaVersionError
        
        self._expand = expand_workflow
        
        self._root_wf_id = None
        self._root_wf_uuid = None
        self._job_filter_mode = None
        self._time_filter_mode = None
        self._host_filter = None
        self._xform_filter = {'include':None, 'exclude':None}
        
        self._wfs = []
        pass
    
    def initialize(self, root_wf_uuid):
        self.log.debug('initialize')
        self._root_wf_uuid = root_wf_uuid
        q = self.session.query(Workflow.wf_id).filter(Workflow.wf_uuid == self._root_wf_uuid)      
        
        try:
            self._root_wf_id = q.one().wf_id
        except orm.exc.MultipleResultsFound, e:
            self.log.error('initialize',
                msg='Multiple results found for wf_uuid: %s' % root_wf_uuid)
            return False
        except orm.exc.NoResultFound, e:
            self.log.error('initialize',
                msg='No results found for wf_uuid: %s' % root_wf_uuid)
            return False
        
        if self._expand:
            q = self.session.query(Workflow.wf_id).filter(Workflow.root_wf_id == self._root_wf_id)
            for row in q.all():
                self._wfs.append(row.wf_id)
            if not len(self._wfs):
                self.log.error('initialize',
                    msg='Unable to expand wf_uuid: %s - not a root_wf_id?' % root_wf_uuid)
                return False
        else:
            self._wfs.append(self._root_wf_id)
            
        if not len(self._wfs):
            self.log.error('initialize',
                msg='No results found for wf_uuid: %s' % root_wf_uuid)
            return False
        
        # Initialize filters with default value
        self.set_job_filter()
        self.set_time_filter()
        self.set_host_filter()
        self.set_transformation_filter()
        return True
        
    def close(self):
        self.log.debug('close')
        self.disconnect()
        
    def set_job_filter(self, filter='all'):
        modes = ['all', 'nonsub', 'subwf', 'dax', 'dag', 'compute', 'stage-in-tx', 
                'stage-out-tx', 'registration', 'inter-site-tx', 'create-dir', 
                'staged-compute', 'cleanup', 'chmod']
        try:
            modes.index(filter)
            self._job_filter_mode = filter
            self.log.debug('set_job_filter', msg='Setting filter to: %s' % filter)
        except:
            self._job_filter_mode = 'all'
            self.log.error('set_job_filter', msg='Unknown job filter %s - setting to all' % filter)
            
            
    def set_time_filter(self, filter='month'):
        modes = ['month', 'week', 'day', 'hour']
        try:
            modes.index(filter)
            self._time_filter_mode = filter
            self.log.debug('set_time_filter', msg='Setting filter to: %s' % filter)
        except:
            self._job_filter_mode = 'month'
            self.log.error('set_time_filter', msg='Unknown time filter %s - setting to month' % filter)
            
    def set_host_filter(self, host=None):
        """
        The host argument can either be a string/single hostname or
        it can be a list/array of hostnames.
        """
        self._host_filter = host
        
    def set_transformation_filter(self, include=None, exclude=None):
        """
        Either of these args can either be a single string/xform type or
        it can be a list/array of xform types.
        
        Both arguments can not be set at the same time.  If they are,
        the program will log an error and not do any filtering.
        """
        self._xform_filter['include'] = include
        self._xform_filter['exclude'] = exclude
                
    #
    # Pulls information about sub workflows
    #
            
    def get_sub_workflow_ids(self):
        """
        Returns info on child workflows only.
        """
        q = self.session.query(Workflow.wf_id, Workflow.wf_uuid, Workflow.dax_label)
        q = q.filter(Workflow.parent_wf_id == self._root_wf_id)
        return q.all()
        
    def get_descendant_workflow_ids(self):
        q = self.session.query(Workflow.wf_id, Workflow.wf_uuid)
        q = q.filter(Workflow.root_wf_id == self._root_wf_id)
        q = q.filter(Workflow.wf_id != self._root_wf_id)
        return q.all()
        
    def get_schema_version(self):
        return self.s_check.check_version()
            
    #
    # Status of initially planned wf components.
    #
    
    #
    # The following block of queries are documented here:
    # https://confluence.pegasus.isi.edu/display/pegasus/Workflow+Summary
    # and
    # https://confluence.pegasus.isi.edu/display/pegasus/Workflow+Statistics+file
    #
    
    def _dax_or_dag_cond(self, JobO=Job):
        return or_(JobO.type_desc == 'dax', JobO.type_desc == 'dag')
        
    def _get_job_filter(self, JobO=Job):
        filters = {
            'all': None,
            'nonsub': not_(self._dax_or_dag_cond(JobO)),
            'subwf': self._dax_or_dag_cond(JobO),
            'dax': JobO.type_desc == 'dax',
            'dag': JobO.type_desc == 'dag',
            'compute': JobO.type_desc == 'compute',
            'stage-in-tx': JobO.type_desc == 'stage-in-tx',
            'stage-out-tx': JobO.type_desc == 'stage-out-tx',
            'registration': JobO.type_desc == 'registration',
            'inter-site-tx': JobO.type_desc == 'inter-site-tx',
            'create-dir': JobO.type_desc == 'create-dir',
            'staged-compute': JobO.type_desc == 'staged-compute',
            'cleanup': JobO.type_desc == 'cleanup',
            'chmod': JobO.type_desc == 'chmod',
        }
        return filters[self._job_filter_mode]
        
    def _max_job_seq_subquery(self):
        """
        Creates the following subquery that is used in 
        several queries:
        and jb_inst.job_submit_seq  = (
            select max(job_submit_seq) from job_instance where job_id = jb_inst.job_id group by job_id
            )
        """
        JobInstanceSubMax = orm.aliased(JobInstance)
        sub_q = self.session.query(func.max(JobInstanceSubMax.job_submit_seq).label('max_id'))
        sub_q = sub_q.filter(JobInstanceSubMax.job_id == JobInstance.job_id).correlate(JobInstance)
        sub_q = sub_q.group_by(JobInstanceSubMax.job_id).subquery()
        return sub_q
        
    def get_total_jobs_status(self):
        """
        https://confluence.pegasus.isi.edu/display/pegasus/Workflow+Summary#WorkflowSummary-Totaljobs
        https://confluence.pegasus.isi.edu/display/pegasus/Workflow+Statistics+file#WorkflowStatisticsfile-Totaljobs        
        """
        q = self.session.query(Job.job_id)
        if self._expand:
            q = q.filter(Workflow.root_wf_id == self._root_wf_id)
        else:
            q = q.filter(Workflow.wf_id == self._wfs[0])
        q = q.filter(Job.wf_id == Workflow.wf_id)
        if self._get_job_filter() is not None:
            q = q.filter(self._get_job_filter())
             
        return q.count()
    
        
    def get_total_succeeded_jobs_status(self):
        """
        https://confluence.pegasus.isi.edu/display/pegasus/Workflow+Summary#WorkflowSummary-Totalsucceededjobs
        https://confluence.pegasus.isi.edu/display/pegasus/Workflow+Statistics+file#WorkflowStatisticsfile-Totalsucceededjobs
        """
        JobInstanceSub = orm.aliased(JobInstance, name='JobInstanceSub')
        sq_1 = self.session.query(func.max(JobInstanceSub.job_submit_seq).label('jss'), JobInstanceSub.job_id.label('jobid'),
                JobInstanceSub.exitcode.label('ec'))
        if self._expand:
            sq_1 = sq_1.filter(Workflow.root_wf_id == self._root_wf_id)
        else:
            sq_1 = sq_1.filter(Workflow.wf_id == self._wfs[0])
        sq_1 = sq_1.filter(Workflow.wf_id == Job.wf_id)
        sq_1 = sq_1.filter(Job.job_id == JobInstanceSub.job_id)
        if self._get_job_filter() is not None:
            sq_1 = sq_1.filter(self._get_job_filter())
        sq_1 = sq_1.group_by(JobInstanceSub.job_id).subquery()
        
        q = self.session.query(JobInstance.job_instance_id.label('last_job_instance'))
        q = q.filter(JobInstance.job_id == sq_1.c.jobid)
        q = q.filter(JobInstance.job_submit_seq == sq_1.c.jss)
        q = q.filter(sq_1.c.ec == 0).filter(sq_1.c.ec != None)
        
        return q.count()
    
        
    def get_total_failed_jobs_status(self):
        """
        https://confluence.pegasus.isi.edu/display/pegasus/Workflow+Summary#WorkflowSummary-Totalfailedjobs
        https://confluence.pegasus.isi.edu/display/pegasus/Workflow+Statistics+file#WorkflowStatisticsfile-Totalfailedjobs
        """
        JobInstanceSub = orm.aliased(JobInstance, name='JobInstanceSub')
        sq_1 = self.session.query(func.max(JobInstanceSub.job_submit_seq).label('jss'), JobInstanceSub.job_id.label('jobid'),
                JobInstanceSub.exitcode.label('ec'))
        if self._expand:
            sq_1 = sq_1.filter(Workflow.root_wf_id == self._root_wf_id)
        else:
            sq_1 = sq_1.filter(Workflow.wf_id == self._wfs[0])
        sq_1 = sq_1.filter(Workflow.wf_id == Job.wf_id)
        sq_1 = sq_1.filter(Job.job_id == JobInstanceSub.job_id)
        if self._get_job_filter() is not None:
            sq_1 = sq_1.filter(self._get_job_filter())
        sq_1 = sq_1.group_by(JobInstanceSub.job_id).subquery()
        
        q = self.session.query(JobInstance.job_instance_id.label('last_job_instance'))
        q = q.filter(JobInstance.job_id == sq_1.c.jobid)
        q = q.filter(JobInstance.job_submit_seq == sq_1.c.jss)
        q = q.filter(sq_1.c.ec != 0).filter(sq_1.c.ec != None)
        
        return q.count()
        
    def _query_jobstate_for_instance(self, states):
        """
        The states arg is a list of strings.
        Returns an appropriate subquery.
        """
        q = self.session.query(Jobstate.job_instance_id)
        q = q.filter(Jobstate.job_instance_id == JobInstance.job_instance_id).correlate(JobInstance)
        q = q.filter(Jobstate.state.in_(states)).subquery()
        return q
        
    def get_total_jobs_retries(self):
        """
        https://confluence.pegasus.isi.edu/display/pegasus/Workflow+Summary#WorkflowSummary-TotalJobRetries
        https://confluence.pegasus.isi.edu/display/pegasus/Workflow+Statistics+file#WorkflowStatisticsfile-TotalJobRetries
        """
        d_or_d = self._dax_or_dag_cond()
        
        sq_1 = self.session.query(func.count(Job.job_id))
        if self._expand:
            sq_1 = sq_1.filter(Workflow.root_wf_id == self._root_wf_id)
        else:
            sq_1 = sq_1.filter(Workflow.wf_id == self._wfs[0])
        sq_1 = sq_1.filter(Job.wf_id == Workflow.wf_id)
        sq_1 = sq_1.filter(Job.job_id == JobInstance.job_id)
        if self._get_job_filter() is not None:
            sq_1 = sq_1.filter(self._get_job_filter())

        sq_1 = sq_1.subquery()

        
        sq_2 = self.session.query(func.count(distinct(JobInstance.job_id)))
        if self._expand:
            sq_2 = sq_2.filter(Workflow.root_wf_id == self._root_wf_id)
        else:
            sq_2 = sq_2.filter(Workflow.wf_id == self._wfs[0])
        sq_2 = sq_2.filter(Job.wf_id == Workflow.wf_id)
        sq_2 = sq_2.filter(Job.job_id == JobInstance.job_id)
        if self._get_job_filter() is not None:
            sq_2 = sq_2.filter(self._get_job_filter())

        sq_2 = sq_2.subquery()
        
        q = self.session.query((sq_1.as_scalar() - sq_2.as_scalar()).label('total_job_retries'))
        
        return q.all()[0].total_job_retries
        
    def get_total_tasks_status(self):
        """
        https://confluence.pegasus.isi.edu/display/pegasus/Workflow+Summary#WorkflowSummary-Totaltask
        https://confluence.pegasus.isi.edu/display/pegasus/Workflow+Statistics+file#WorkflowStatisticsfile-Totaltasks
        """
        q = self.session.query(Task.task_id)
        if self._expand:
            q = q.filter(Workflow.root_wf_id == self._root_wf_id)
        else:
            q = q.filter(Workflow.wf_id == self._wfs[0])
        q = q.filter(Task.wf_id == Workflow.wf_id)
        q = q.filter(Task.job_id == Job.job_id)
        if self._get_job_filter(Task) is not None:
            q = q.filter(self._get_job_filter(Task))
        return q.count()
    
        
    def _base_task_status_query_old(self):
        """
        https://confluence.pegasus.isi.edu/display/pegasus/Workflow+Summary#WorkflowSummary-Totalsucceededtasks
        https://confluence.pegasus.isi.edu/display/pegasus/Workflow+Statistics+file#WorkflowStatisticsfile-Totalsucceededtasks
        """
        # This query generation method is obsolete and is only being
        # kept for optimization reference.
        
        WorkflowSub1 = orm.aliased(Workflow, name='WorkflowSub1')
        JobInstanceSub1 = orm.aliased(JobInstance, name='JobInstanceSub1')
        JobSub1 = orm.aliased(Job, name='JobSub1')
        
        sq_1 = self.session.query(WorkflowSub1.wf_id.label('wid'), 
                func.max(JobInstanceSub1.job_submit_seq).label('jss'),
                JobInstanceSub1.job_id.label('jobid')
        )
        if self._expand:
            sq_1 = sq_1.filter(WorkflowSub1.root_wf_id == self._root_wf_id)
        else:
            sq_1 = sq_1.filter(WorkflowSub1.wf_id == self._wfs[0])
        sq_1 = sq_1.filter(WorkflowSub1.wf_id == JobSub1.wf_id)
        sq_1 = sq_1.filter(JobSub1.job_id == JobInstanceSub1.job_id)
        sq_1 = sq_1.group_by(JobInstanceSub1.job_id)
        if self._get_job_filter(JobSub1) is not None:
            sq_1 = sq_1.filter(self._get_job_filter(JobSub1))
        sq_1 = sq_1.subquery()
        
        JobInstanceSub2 = orm.aliased(JobInstance, name='JobInstanceSub2')
        sq_2 = self.session.query(sq_1.c.wid.label('wf_id'), JobInstanceSub2.job_instance_id.label('last_job_instance_id'))
        sq_2 = sq_2.filter(JobInstanceSub2.job_id == sq_1.c.jobid)
        sq_2 = sq_2.filter(JobInstanceSub2.job_submit_seq == sq_1.c.jss)
        sq_2 = sq_2.subquery()
        
        q = self.session.query(Invocation.invocation_id)
        q = q.filter(Invocation.abs_task_id != None)
        q = q.filter(Invocation.job_instance_id == sq_2.c.last_job_instance_id)
        q = q.filter(Invocation.wf_id == sq_2.c.wf_id)
        
        # Calling wrapper methods would invoke like so:
        # q = self._base_task_status_query()
        # q = q.filter(Invocation.exitcode == 0)
        # return q.count()
        
        return q
        
    def _base_task_statistics_query(self, success=True):
        w = orm.aliased(Workflow, name='w')
        j = orm.aliased(Job, name='j')
        ji = orm.aliased(JobInstance, name='ji')
        tk = orm.aliased(Task, name='tk')

        sq_1 = self.session.query(w.wf_id, 
                j.job_id,
                ji.job_instance_id.label('jiid'),
                ji.job_submit_seq.label('jss'),
                func.max(ji.job_submit_seq).label('maxjss'))
        sq_1 = sq_1.join(j, w.wf_id == j.wf_id)
        sq_1 = sq_1.join(ji, j.job_id == ji.job_id)
        if self._expand:
            sq_1 = sq_1.filter(w.root_wf_id == self._root_wf_id)
        else:
            sq_1 = sq_1.filter(w.wf_id == self._wfs[0])
        sq_1 = sq_1.group_by(j.job_id)
        sq_1 = sq_1.subquery('t')

        sq_2 = self.session.query(sq_1.c.wf_id, func.count(Invocation.exitcode).label('count'))
        sq_2 = sq_2.select_from(orm.join(sq_1, Invocation, sq_1.c.jiid == Invocation.job_instance_id))
        sq_2 = sq_2.join(tk, tk.abs_task_id == Invocation.abs_task_id)
        sq_2 = sq_2.filter(tk.type_desc != 'dax')
        sq_2 = sq_2.filter(sq_1.c.jss == sq_1.c.maxjss)
        sq_2 = sq_2.filter(Invocation.abs_task_id != None)
        if success:
            sq_2 = sq_2.filter(Invocation.exitcode == 0)
        else:
            sq_2 = sq_2.filter(Invocation.exitcode != 0)
        sq_2 = sq_2.group_by(sq_1.c.wf_id)
        
        return sq_2
        
    def _task_statistics_query_sum(self, success=True):
        s = self._base_task_statistics_query(success).subquery('tt')
        q = self.session.query(func.sum(s.c.count).label('task_count'))
        return q.one()[0] or 0
    
    def get_total_succeeded_tasks_status(self):
        return self._task_statistics_query_sum(success=True)

    def get_total_failed_tasks_status(self):
{'fraction_non_alphanumeric': 0.09928428035262285, 'fraction_numerical': 0.008379156847342237, 'mean_word_length': 2.689638515417438, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 30, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2405457', 'n_tokens_mistral': 15110, 'n_tokens_neox': 14567, 'n_words': 3164}