Index: /trunk/jobmond/jobmond.py
===================================================================
--- /trunk/jobmond/jobmond.py	(revision 662)
+++ /trunk/jobmond/jobmond.py	(revision 663)
@@ -3,6 +3,5 @@
 # This file is part of Jobmonarch
 #
-# Copyright (C) 2006-2007  Ramon Bastiaans
-# Copyright (C) 2007, 2009  Dave Love  (SGE code)
+# Copyright (C) 2006-2012  Ramon Bastiaans
 # 
 # Jobmonarch is free software; you can redistribute it and/or modify
@@ -25,9 +24,7 @@
 import sys, getopt, ConfigParser, time, os, socket, string, re
 import xdrlib, socket, syslog, xml, xml.sax
-from xml.sax.handler import feature_namespaces
-from collections import deque
 from types import *
 
-VERSION='0.3.1'
+VERSION='TRUNK+SVN'
 
 def usage( ver ):
@@ -719,165 +716,4 @@
             time.sleep( BATCH_POLL_INTERVAL )   
 
-# SGE code by Dave Love <fx@gnu.org>.  Tested with SGE 6.0u8 and 6.0u11.  May
-# work with SGE 6.1 (else should be easily fixable), but definitely doesn't
-# with 6.2.  See also the fixmes.
-
-class NoJobs (Exception):
-    """Exception raised by empty job list in qstat output."""
-    pass
-
-class SgeQstatXMLParser(xml.sax.handler.ContentHandler):
-    """SAX handler for XML output from Sun Grid Engine's `qstat'."""
-
-    def __init__(self):
-        self.value = ""
-        self.joblist = []
-        self.job = {}
-        self.queue = ""
-        self.in_joblist = False
-        self.lrequest = False
-        self.eltq = deque()
-        xml.sax.handler.ContentHandler.__init__(self)
-
-    # The structure of the output is as follows (for SGE 6.0).  It's
-    # similar for 6.1, but radically different for 6.2, and is
-    # undocumented generally.  Unfortunately it's voluminous, and probably
-    # doesn't scale to large clusters/queues.
-
-    # <detailed_job_info  xmlns:xsd="http://www.w3.org/2001/XMLSchema">
-    #   <djob_info>
-    #     <qmaster_response>  <!-- job -->
-    #       ...
-    #       <JB_ja_template>  
-    #         <ulong_sublist>
-    #         ...             <!-- start_time, state ... -->
-    #         </ulong_sublist>
-    #       </JB_ja_template>  
-    #       <JB_ja_tasks>
-    #         <ulong_sublist>
-    #           ...           <!-- task info
-    #         </ulong_sublist>
-    #         ...
-    #       </JB_ja_tasks>
-    #       ...
-    #     </qmaster_response>
-    #   </djob_info>
-    #   <messages>
-    #   ...
-
-    # NB.  We might treat each task as a separate job, like
-    # straight qstat output, but the web interface expects jobs to
-    # be identified by integers, not, say, <job number>.<task>.
-
-    # So, I lied.  If the job list is empty, we get invalid XML
-    # like this, which we need to defend against:
-
-    # <unknown_jobs  xmlns:xsd="http://www.w3.org/2001/XMLSchema">
-    #   <>
-    #     <ST_name>*</ST_name>
-    #   </>
-    # </unknown_jobs>
-
-    def startElement(self, name, attrs):
-        self.value = ""
-        if name == "djob_info": # job list
-            self.in_joblist = True
-        # The job container is "qmaster_response" in SGE 6.0
-        # and 6.1, but "element" in 6.2.  This is only the very
-        # start of what's necessary for 6.2, though (sigh).
-        elif (name == "qmaster_response" or name == "element") \
-                and self.eltq[-1] == "djob_info": # job
-            self.job = {"job_state": "U", "slots": 0,
-                    "nodes": [], "queued_timestamp": "",
-                    "queued_timestamp": "", "queue": "",
-                    "ppn": "0", "RN_max": 0,
-                    # fixme in endElement
-                    "requested_memory": 0, "requested_time": 0
-                    }
-            self.joblist.append(self.job)
-        elif name == "qstat_l_requests": # resource request
-            self.lrequest = True
-        elif name == "unknown_jobs":
-            raise NoJobs
-        self.eltq.append (name)
-
-    def characters(self, ch):
-        self.value += ch
-
-    def endElement(self, name): 
-        """Snarf job elements contents into job dictionary.
-           Translate keys if appropriate."""
-
-        name_trans = {
-          "JB_job_number": "number",
-          "JB_job_name": "name", "JB_owner": "owner",
-          "queue_name": "queue", "JAT_start_time": "start_timestamp",
-          "JB_submission_time": "queued_timestamp"
-          }
-        value = self.value
-        self.eltq.pop ()
-
-        if name == "djob_info":
-            self.in_joblist = False
-            self.job = {}
-        elif name == "JAT_master_queue":
-            self.job["queue"] = value.split("@")[0]
-        elif name == "JG_qhostname":
-            if not (value in self.job["nodes"]):
-                self.job["nodes"].append(value)
-        elif name == "JG_slots": # slots in use
-            self.job["slots"] += int(value)
-        elif name == "RN_max": # requested slots (tasks or parallel)
-            self.job["RN_max"] = max (self.job["RN_max"],
-                          int(value))
-        elif name == "JAT_state": # job state (bitwise or)
-            value = int (value)
-            # Status values from sge_jobL.h
-            #define JIDLE                   0x00000000
-            #define JHELD                   0x00000010
-            #define JMIGRATING              0x00000020
-            #define JQUEUED                 0x00000040
-            #define JRUNNING                0x00000080
-            #define JSUSPENDED              0x00000100
-            #define JTRANSFERING            0x00000200
-            #define JDELETED                0x00000400
-            #define JWAITING                0x00000800
-            #define JEXITING                0x00001000
-            #define JWRITTEN                0x00002000
-            #define JSUSPENDED_ON_THRESHOLD 0x00010000
-            #define JFINISHED               0x00010000
-            if value & 0x80:
-                self.job["status"] = "R"
-            elif value & 0x40:
-                self.job["status"] = "Q"
-            else:
-                self.job["status"] = "O" # `other'
-        elif name == "CE_name" and self.lrequest and self.value in \
-                ("h_cpu", "s_cpu", "cpu", "h_core", "s_core"):
-            # We're in a container for an interesting resource
-            # request; record which type.
-            self.lrequest = self.value
-        elif name == "CE_doubleval" and self.lrequest:
-            # if we're in a container for an interesting
-            # resource request, use the maxmimum of the hard
-            # and soft requests to record the requested CPU
-            # or core.  Fixme:  I'm not sure if this logic is
-            # right.
-            if self.lrequest in ("h_core", "s_core"):
-                self.job["requested_memory"] = \
-                    max (float (value),
-                     self.job["requested_memory"])
-            # Fixme:  Check what cpu means, c.f [hs]_cpu.
-            elif self.lrequest in ("h_cpu", "s_cpu", "cpu"):
-                self.job["requested_time"] = \
-                    max (float (value),
-                     self.job["requested_time"])
-        elif name == "qstat_l_requests":
-            self.lrequest = False
-        elif self.job and self.in_joblist:
-            if name in name_trans:
-                name = name_trans[name]
-                self.job[name] = value
-
 # Abstracted from PBS original.
 #
@@ -914,273 +750,4 @@
     return nodeslist
 
-class SgeDataGatherer(DataGatherer):
-
-    jobs = {}
-
-    def __init__( self ):
-        self.jobs = {}
-        self.timeoffset = 0
-        self.dp = DataProcessor()
-
-    def getJobData( self ):
-        """Gather all data on current jobs in SGE"""
-
-        import popen2
-
-        self.cur_time = 0
-        queues = ""
-        if QUEUE:   # only for specific queues
-            # Fixme:  assumes queue names don't contain single
-            # quote or comma.  Don't know what the SGE rules are.
-            queues = " -q '" + string.join (QUEUE, ",") + "'"
-        # Note the comment in SgeQstatXMLParser about scaling with
-        # this method of getting data.  I haven't found better one.
-        # Output with args `-xml -ext -f -r' is easier to parse
-        # in some ways, harder in others, but it doesn't provide
-        # the submission time (at least SGE 6.0).  The pipeline
-        # into sed corrects bogus XML observed with a configuration
-        # of SGE 6.0u8, which otherwise causes the parsing to hang.
-        piping = popen2.Popen3("qstat -u '*' -j '*' -xml | \
-sed -e 's/reported usage>/reported_usage>/g' -e 's;<\/*JATASK:.*>;;'" \
-                           + queues, True)
-        qstatparser = SgeQstatXMLParser()
-        parse_err = 0
-        try:
-            xml.sax.parse(piping.fromchild, qstatparser)
-        except NoJobs:
-            pass
-        except:
-            parse_err = 1
-            if piping.wait():
-                debug_msg(10,
-                  "qstat error, skipping until next polling interval: "
-                  + piping.childerr.readline())
-                return None
-            elif parse_err:
-                debug_msg(10, "Bad XML output from qstat"())
-                exit (1)
-        for f in piping.fromchild, piping.tochild, piping.childerr:
-            f.close()
-        self.cur_time = time.time()
-        jobs_processed = []
-        for job in qstatparser.joblist:
-            job_id = job["number"]
-            if job["status"] in [ 'Q', 'R' ]:
-                jobs_processed.append(job_id)
-            if job["status"] == "R":
-                job["nodes"] = do_nodelist (job["nodes"])
-                # Fixme: why is job["nodes"] sometimes null?
-                try:
-                    # Fixme: Is this sensible?  The
-                    # PBS-type PPN isn't something you use
-                    # with SGE.
-                    job["ppn"] = float(job["slots"]) / \
-                        len(job["nodes"])
-                except:
-                    job["ppn"] = 0
-                if DETECT_TIME_DIFFS:
-                    # If a job start is later than our
-                    # current date, that must mean
-                    # the SGE server's time is later
-                    # than our local time.
-                    start_timestamp = \
-                        int (job["start_timestamp"])
-                    if start_timestamp > \
-                            int(self.cur_time) + \
-                            int(self.timeoffset):
-
-                        self.timeoffset = \
-                            start_timestamp - \
-                            int(self.cur_time)
-            else:
-                # fixme: Note sure what this should be:
-                job["ppn"] = job["RN_max"]
-                job["nodes"] = "1"
-
-            myAttrs = {}
-            for attr in ["name", "queue", "owner",
-                     "requested_time", "status",
-                     "requested_memory", "ppn",
-                     "start_timestamp", "queued_timestamp"]:
-                myAttrs[attr] = str(job[attr])
-            myAttrs["nodes"] = job["nodes"]
-            myAttrs["reported"] = str(int(self.cur_time) + \
-                          int(self.timeoffset))
-            myAttrs["domain"] = fqdn_parts(socket.getfqdn())[1]
-            myAttrs["poll_interval"] = str(BATCH_POLL_INTERVAL)
-
-            if self.jobDataChanged(self.jobs, job_id, myAttrs) \
-                    and myAttrs["status"] in ["R", "Q"]:
-                self.jobs[job_id] = myAttrs
-        for id, attrs in self.jobs.items():
-            if id not in jobs_processed:
-                del self.jobs[id]
-
-# LSF code by Mahmoud Hanafi <hanafim@users.sourceforge.nt>
-# Requres LSFObject http://sourceforge.net/projects/lsfobject
-#
-class LsfDataGatherer(DataGatherer):
-
-        """This is the DataGatherer for LSf"""
-
-        global lsfObject
-
-        def __init__( self ):
-
-                self.jobs = { }
-                self.timeoffset = 0
-                self.dp = DataProcessor()
-                self.initLsfQuery()
-
-        def _countDuplicatesInList( self, dupedList ):
-
-            countDupes  = { }
-
-            for item in dupedList:
-
-                if not countDupes.has_key( item ):
-
-                    countDupes[ item ]  = 1
-                else:
-                    countDupes[ item ]  = countDupes[ item ] + 1
-
-            dupeCountList   = [ ]
-
-            for item, count in countDupes.items():
-
-                dupeCountList.append( ( item, count ) )
-
-            return dupeCountList
-#
-#lst = ['I1','I2','I1','I3','I4','I4','I7','I7','I7','I7','I7']
-#print _countDuplicatesInList(lst)
-#[('I1', 2), ('I3', 1), ('I2', 1), ('I4', 2), ('I7', 5)]
-########################
-
-        def initLsfQuery( self ):
-                self.pq = None
-                self.pq = lsfObject.jobInfoEntObject()
-
-        def getJobData( self, known_jobs="" ):
-                """Gather all data on current jobs in LSF"""
-                if len( known_jobs ) > 0:
-                        jobs = known_jobs
-                else:
-                        jobs = { }
-                joblist = {}
-                joblist = self.pq.getJobInfo()
-                nodelist = ''
-
-                self.cur_time = time.time()
-
-                jobs_processed = [ ]
-
-                for name, attrs in joblist.items():
-                        job_id = str(name)
-                        jobs_processed.append( job_id )
-                        name = self.getAttr( attrs, 'jobName' )
-                        queue = self.getAttr( self.getAttr( attrs, 'submit') , 'queue' )
-                        owner = self.getAttr( attrs, 'user' )
-
-### THIS IS THE rLimit List index values
-#define LSF_RLIMIT_CPU      0            /* cpu time in milliseconds */
-#define LSF_RLIMIT_FSIZE    1            /* maximum file size */
-#define LSF_RLIMIT_DATA     2            /* data size */
-#define LSF_RLIMIT_STACK    3            /* stack size */
-#define LSF_RLIMIT_CORE     4            /* core file size */
-#define LSF_RLIMIT_RSS      5            /* resident set size */
-#define LSF_RLIMIT_NOFILE   6            /* open files */
-#define LSF_RLIMIT_OPEN_MAX 7            /* (from HP-UX) */
-#define LSF_RLIMIT_VMEM     8            /* maximum swap mem */
-#define LSF_RLIMIT_SWAP     8
-#define LSF_RLIMIT_RUN      9            /* max wall-clock time limit */
-#define LSF_RLIMIT_PROCESS  10           /* process number limit */
-#define LSF_RLIMIT_THREAD   11           /* thread number limit (introduced in LSF6.0) */
-#define LSF_RLIM_NLIMITS    12           /* number of resource limits */
-
-                        requested_time = self.getAttr( self.getAttr( attrs, 'submit') , 'rLimits' )[9]
-                        if requested_time == -1: 
-                                requested_time = ""
-                        requested_memory = self.getAttr( self.getAttr( attrs, 'submit') , 'rLimits' )[8]
-                        if requested_memory == -1: 
-                                requested_memory = ""
-# This tries to get proc per node. We don't support this right now
-                        ppn = 0 #self.getAttr( self.getAttr( attrs, 'SubmitList') , 'numProessors' )
-                        requested_cpus = self.getAttr( self.getAttr( attrs, 'submit') , 'numProcessors' )
-                        if requested_cpus == None or requested_cpus == "":
-                                requested_cpus = 1
-
-                        if QUEUE:
-                            for q in QUEUE:
-                                if q == queue:
-                                    display_queue = 1
-                                    break
-                                else:
-                                    display_queue = 0
-                                    continue
-                        if display_queue == 0:
-                            continue
-
-                        runState = self.getAttr( attrs, 'status' )
-                        if runState == 4:
-                                status = 'R'
-                        else:
-                                status = 'Q'
-                        queued_timestamp = self.getAttr( attrs, 'submitTime' )
-
-                        if status == 'R':
-                                start_timestamp = self.getAttr( attrs, 'startTime' )
-                                nodesCpu =  dict(self._countDuplicatesInList(self.getAttr( attrs, 'exHosts' )))
-                                nodelist = nodesCpu.keys()
-
-                                if DETECT_TIME_DIFFS:
-
-                                        # If a job start if later than our current date,
-                                        # that must mean the Torque server's time is later
-                                        # than our local time.
-
-                                        if int(start_timestamp) > int( int(self.cur_time) + int(self.timeoffset) ):
-
-                                                self.timeoffset = int( int(start_timestamp) - int(self.cur_time) )
-
-                        elif status == 'Q':
-                                start_timestamp = ''
-                                count_mynodes = 0
-                                numeric_node = 1
-                                nodelist = ''
-
-                        myAttrs = { }
-                        if name == "":
-                                myAttrs['name'] = "none"
-                        else:
-                                myAttrs['name'] = name
-
-                        myAttrs[ 'owner' ]      = owner
-                        myAttrs[ 'requested_time' ] = str(requested_time)
-                        myAttrs[ 'requested_memory' ]   = str(requested_memory)
-                        myAttrs[ 'requested_cpus' ] = str(requested_cpus)
-                        myAttrs[ 'ppn' ]        = str( ppn )
-                        myAttrs[ 'status' ]     = status
-                        myAttrs[ 'start_timestamp' ]    = str(start_timestamp)
-                        myAttrs[ 'queue' ]      = str(queue)
-                        myAttrs[ 'queued_timestamp' ]   = str(queued_timestamp)
-                        myAttrs[ 'reported' ]       = str( int( int( self.cur_time ) + int( self.timeoffset ) ) )
-                        myAttrs[ 'nodes' ]      = do_nodelist( nodelist )
-                        myAttrs[ 'domain' ]     = fqdn_parts( socket.getfqdn() )[1]
-                        myAttrs[ 'poll_interval' ]  = str(BATCH_POLL_INTERVAL)
-
-                        if self.jobDataChanged( jobs, job_id, myAttrs ) and myAttrs['status'] in [ 'R', 'Q' ]:
-                                jobs[ job_id ] = myAttrs
-
-                                debug_msg( 10, printTime() + ' job %s state changed' %(job_id) )
-
-                for id, attrs in jobs.items():
-                        if id not in jobs_processed:
-                                # This one isn't there anymore
-                                #
-                                del jobs[ id ]
-
-                self.jobs = jobs
-
 class PbsDataGatherer( DataGatherer ):
 
@@ -1235,5 +802,4 @@
             queue           = self.getAttr( attrs, 'queue' )
             nodect          = self.getAttr( attrs['Resource_List'], 'nodect' )
-            exec_group      = self.getAttr( attrs, 'egroup' )
 
             requested_time      = self.getAttr( attrs['Resource_List'], 'walltime' )
@@ -1374,5 +940,4 @@
             myAttrs[ 'owner' ]             = str( owner )
             myAttrs[ 'nodect' ]            = str( nodect )
-            myAttrs[ 'exec.group' ]        = str( exec_group )
             myAttrs[ 'exec.hostnames' ]    = str( running_nodes )
             myAttrs[ 'exec.nodestr' ]      = str( exec_nodestr )
