source: branches/1.0/jobmond/jobmond.py @ 903

Last change on this file since 903 was 903, checked in by ramonb, 8 years ago

jobmond/jobmond.py:

  • lowered gmetric version from 3.4.0 to 3.3.8
  • Property svn:executable set to *
  • Property svn:keywords set to Id
File size: 64.5 KB
Line 
1#!/usr/bin/env python
2#
3# This file is part of Jobmonarch
4#
5# Copyright (C) 2006-2013  Ramon Bastiaans
6# Copyright (C) 2007, 2009  Dave Love  (SGE code)
7#
8# Jobmonarch is free software; you can redistribute it and/or modify
9# it under the terms of the GNU General Public License as published by
10# the Free Software Foundation; either version 2 of the License, or
11# (at your option) any later version.
12#
13# Jobmonarch is distributed in the hope that it will be useful,
14# but WITHOUT ANY WARRANTY; without even the implied warranty of
15# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
16# GNU General Public License for more details.
17#
18# You should have received a copy of the GNU General Public License
19# along with this program; if not, write to the Free Software
20# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
21#
22# SVN $Id: jobmond.py 903 2013-05-22 11:39:52Z ramonb $
23#
24
25# vi :set ts=4
26
27import sys, getopt, ConfigParser, time, os, socket, string, re
28import xdrlib, socket, syslog, xml, xml.sax, shlex, os.path, pwd
29from xml.sax.handler import feature_namespaces
30from collections import deque
31from glob import glob
32
33VERSION='1.1'
34
35def usage( ver ):
36
37    print 'jobmond %s' %VERSION
38
39    if ver:
40        return 0
41
42    print
43    print 'Purpose:'
44    print '  The Job Monitoring Daemon (jobmond) reports batch jobs information and statistics'
45    print '  to Ganglia, which can be viewed with Job Monarch web frontend'
46    print
47    print 'Usage:    jobmond [OPTIONS]'
48    print
49    print '  -c, --config=FILE    The configuration file to use (default: /etc/jobmond.conf)'
50    print '  -p, --pidfile=FILE    Use pid file to store the process id'
51    print '  -h, --help        Print help and exit'
52    print '  -v, --version      Print version and exit'
53    print
54
55def processArgs( args ):
56
57    SHORT_L      = 'p:hvc:'
58    LONG_L       = [ 'help', 'config=', 'pidfile=', 'version' ]
59
60    global PIDFILE, JOBMOND_CONF
61    PIDFILE      = None
62
63    JOBMOND_CONF = '/etc/jobmond.conf'
64
65    try:
66
67        opts, args    = getopt.getopt( args, SHORT_L, LONG_L )
68
69    except getopt.GetoptError, detail:
70
71        print detail
72        usage( False )
73        sys.exit( 1 )
74
75    for opt, value in opts:
76
77        if opt in [ '--config', '-c' ]:
78       
79            JOBMOND_CONF = value
80
81        if opt in [ '--pidfile', '-p' ]:
82
83            PIDFILE      = value
84       
85        if opt in [ '--help', '-h' ]:
86 
87            usage( False )
88            sys.exit( 0 )
89
90        if opt in [ '--version', '-v' ]:
91
92            usage( True )
93            sys.exit( 0 )
94
95    return loadConfig( JOBMOND_CONF )
96
97class GangliaConfigParser:
98
99    def __init__( self, filename ):
100
101        self.conf_lijst   = [ ]
102        self.conf_dict    = { }
103        self.filename     = filename
104        self.file_pointer = file( filename, 'r' )
105        self.lexx         = shlex.shlex( self.file_pointer )
106        self.lexx.whitespace_split = True
107
108        self.parse()
109
110    def __del__( self ):
111
112        """
113        Cleanup: close file descriptor
114        """
115
116        self.file_pointer.close()
117        del self.lexx
118        del self.conf_lijst
119
120    def removeQuotes( self, value ):
121
122        clean_value = value
123        clean_value = clean_value.replace( "'", "" )
124        clean_value = clean_value.replace( '"', '' )
125        clean_value = clean_value.strip()
126
127        return clean_value
128
129    def removeBraces( self, value ):
130
131        clean_value = value
132        clean_value = clean_value.replace( "(", "" )
133        clean_value = clean_value.replace( ')', '' )
134        clean_value = clean_value.strip()
135
136        return clean_value
137
138    def parse( self ):
139
140        """
141        Parse self.filename using shlex scanning.
142        - Removes /* comments */
143        - Traverses (recursively) through all include () statements
144        - Stores complete valid config tokens in self.conf_list
145
146        i.e.:
147            ['globals',
148             '{',
149             'daemonize',
150             '=',
151             'yes',
152             'setuid',
153             '=',
154             'yes',
155             'user',
156             '=',
157             'ganglia',
158             'debug_level',
159             '=',
160             '0',
161             <etc> ]
162        """
163
164        t = 'bogus'
165        c = False
166        i = False
167
168        while t != self.lexx.eof:
169            #print 'get token'
170            t = self.lexx.get_token()
171
172            if len( t ) >= 2:
173
174                if len( t ) >= 4:
175
176                    if t[:2] == '/*' and t[-2:] == '*/':
177
178                        #print 'comment line'
179                        #print 'skipping: %s' %t
180                        continue
181
182                if t == '/*' or t[:2] == '/*':
183                    c = True
184                    #print 'comment start'
185                    #print 'skipping: %s' %t
186                    continue
187
188                if t == '*/' or t[-2:] == '*/':
189                    c = False
190                    #print 'skipping: %s' %t
191                    #print 'comment end'
192                    continue
193
194            if c:
195                #print 'skipping: %s' %t
196                continue
197
198            if t == 'include':
199                i = True
200                #print 'include start'
201                #print 'skipping: %s' %t
202                continue
203
204            if i:
205
206                #print 'include start: %s' %t
207
208                t2 = self.removeQuotes( t )
209                t2 = self.removeBraces( t )
210
211                for in_file in glob( self.removeQuotes(t2) ):
212
213                    #print 'including file: %s' %in_file
214                    parse_infile = GangliaConfigParser( in_file )
215
216                    self.conf_lijst = self.conf_lijst + parse_infile.getConfLijst()
217
218                    del parse_infile
219
220                i = False
221                #print 'include end'
222                #print 'skipping: %s' %t
223                continue
224
225            #print 'keep: %s' %t
226            self.conf_lijst.append( self.removeQuotes(t) )
227
228    def getConfLijst( self ):
229
230        return self.conf_lijst
231
232    def confListToDict( self, parent_list=None ):
233
234        """
235        Recursively traverses a conf_list and creates dictionary from it
236        """
237
238        new_dict = { }
239        count    = 0
240        skip     = 0
241
242        if not parent_list:
243            parent_list = self.conf_lijst
244
245        #print 'entering confListToDict(): (parent) list size %s' %len(parent_list)
246
247        for n, c in enumerate( parent_list ):
248
249            count = count + 1
250
251            #print 'CL: n %d c %s' %(n, c)
252
253            if skip > 0:
254
255                #print '- skipped'
256                skip = skip - 1
257                continue
258
259            if (n+1) <= (len( parent_list )-1):
260
261                if parent_list[(n+1)] == '{':
262
263                    if not new_dict.has_key( c ):
264                        new_dict[ c ] = [ ]
265
266                    (temp_new_dict, skip) = self.confListToDict( parent_list[(n+2):] )
267                    new_dict[ c ].append( temp_new_dict )
268
269                if parent_list[(n+1)] == '=' and (n+2) <= (len( parent_list )-1):
270
271                    if not new_dict.has_key( c ):
272                        new_dict[ c ] = [ ]
273
274                    new_dict[ c ].append( parent_list[ (n+2) ] )
275
276                    skip = 2
277
278                if parent_list[n] == '}':
279
280                    #print 'leaving confListToDict(): new dict = %s' %new_dict
281                    return (new_dict, count)
282
283    def makeConfDict( self ):
284
285        """
286        Walks through self.conf_list and creates a dictionary based upon config values
287
288        i.e.:
289            'tcp_accept_channel': [{'acl': [{'access': [{'action': ['"allow"'],
290                                                         'ip': ['"127.0.0.1"'],
291                                                         'mask': ['32']}]}],
292                                    'port': ['8649']}],
293            'udp_recv_channel': [{'port': ['8649']}],
294            'udp_send_channel': [{'host': ['145.101.32.3'],
295                                  'port': ['8649']},
296                                 {'host': ['145.101.32.207'],
297                                  'port': ['8649']}]}
298        """
299
300        new_dict = { }
301        skip     = 0
302
303        #print 'entering makeConfDict()'
304
305        for n, c in enumerate( self.conf_lijst ):
306
307            #print 'M: n %d c %s' %(n, c)
308
309            if skip > 0:
310
311                #print '- skipped'
312                skip = skip - 1
313                continue
314
315            if (n+1) <= (len( self.conf_lijst )-1):
316
317                if self.conf_lijst[(n+1)] == '{':
318
319                    if not new_dict.has_key( c ):
320                        new_dict[ c ] = [ ]
321
322                    ( temp_new_dict, skip ) = self.confListToDict( self.conf_lijst[(n+2):] )
323                    new_dict[ c ].append( temp_new_dict )
324
325                if self.conf_lijst[(n+1)] == '=' and (n+2) <= (len( self.conf_lijst )-1):
326
327                    if not new_dict.has_key( c ):
328                        new_dict[ c ] = [ ]
329
330                    new_dict[ c ].append( self.conf_lijst[ (n+2) ] )
331
332                    skip = 2
333
334        self.conf_dict = new_dict
335        #print 'leaving makeConfDict(): conf dict size %d' %len( self.conf_dict )
336
337    def checkConfDict( self ):
338
339        if len( self.conf_lijst ) == 0:
340
341            raise Exception("Something went wrong generating conf list for %s" %self.file_name )
342
343        if len( self.conf_dict ) == 0:
344
345            self.makeConfDict()
346
347    def getConfDict( self ):
348
349        self.checkConfDict()
350        return self.conf_dict
351
352    def getUdpSendChannels( self ):
353
354        self.checkConfDict()
355
356        udp_send_channels = [ ] # IP:PORT
357
358        if not self.conf_dict.has_key( 'udp_send_channel' ):
359            return None
360
361        for u in self.conf_dict[ 'udp_send_channel' ]:
362
363            if u.has_key( 'mcast_join' ):
364
365                ip = u['mcast_join'][0]
366
367            elif u.has_key( 'host' ):
368
369                ip = u['host'][0]
370
371            port = u['port'][0]
372
373            udp_send_channels.append( ( ip, port ) )
374
375        if len( udp_send_channels ) == 0:
376            return None
377
378        return udp_send_channels
379
380    def getSectionLastOption( self, section, option ):
381
382        """
383        Get last option set in a config section that could be set multiple times in multiple (include) files.
384
385        i.e.: getSectionLastOption( 'globals', 'send_metadata_interval' )
386        """
387
388        self.checkConfDict()
389        value = None
390
391        if not self.conf_dict.has_key( section ):
392
393            return None
394
395        # Could be set multiple times in multiple (include) files: get last one set
396        for c in self.conf_dict[ section ]:
397
398                if c.has_key( option ):
399
400                    value = c[ option ][0]
401
402        return value
403
404    def getClusterName( self ):
405
406        return self.getSectionLastOption( 'cluster', 'name' )
407
408    def getVal( self, section, option ):
409
410        return self.getSectionLastOption( section, option )
411
412    def getInt( self, section, valname ):
413
414        value    = self.getVal( section, valname )
415
416        if not value:
417            return None
418
419        return int( value )
420
421    def getStr( self, section, valname ):
422
423        value    = self.getVal( section, valname )
424
425        if not value:
426            return None
427
428        return str( value )
429
430def findGmetric():
431
432    for dir in os.path.expandvars( '$PATH' ).split( ':' ):
433
434        guess    = '%s/%s' %( dir, 'gmetric' )
435
436        if os.path.exists( guess ):
437
438            return guess
439
440    return False
441
442def loadConfig( filename ):
443
444    def getlist( cfg_string ):
445
446        my_list = [ ]
447
448        for item_txt in cfg_string.split( ',' ):
449
450            sep_char = None
451
452            item_txt = item_txt.strip()
453
454            for s_char in [ "'", '"' ]:
455
456                if item_txt.find( s_char ) != -1:
457
458                    if item_txt.count( s_char ) != 2:
459
460                        print 'Missing quote: %s' %item_txt
461                        sys.exit( 1 )
462
463                    else:
464
465                        sep_char = s_char
466                        break
467
468            if sep_char:
469
470                item_txt = item_txt.split( sep_char )[1]
471
472            my_list.append( item_txt )
473
474        return my_list
475
476    if not os.path.isfile( JOBMOND_CONF ):
477
478        print "Is not a file or does not exist: '%s'" %JOBMOND_CONF
479        sys.exit( 1 )
480
481    try:
482        f = open( JOBMOND_CONF, 'r' )
483    except IOError, detail:
484        print "Cannot read config file: '%s'" %JOBMOND_CONF
485        sys.exit( 1 )
486    else:
487        f.close()
488
489    cfg        = ConfigParser.ConfigParser()
490
491    cfg.read( filename )
492
493    global DEBUG_LEVEL, DAEMONIZE, BATCH_SERVER, BATCH_POLL_INTERVAL
494    global GMOND_CONF, DETECT_TIME_DIFFS, BATCH_HOST_TRANSLATE
495    global BATCH_API, QUEUE, GMETRIC_TARGET, USE_SYSLOG
496    global SYSLOG_LEVEL, SYSLOG_FACILITY, GMETRIC_BINARY
497    global METRIC_MAX_VAL_LEN, GMOND_UDP_SEND_CHANNELS
498
499    DEBUG_LEVEL = cfg.getint( 'DEFAULT', 'DEBUG_LEVEL' )
500
501    DAEMONIZE   = cfg.getboolean( 'DEFAULT', 'DAEMONIZE' )
502
503    SYSLOG_LEVEL    = -1
504    SYSLOG_FACILITY = None
505
506    try:
507        USE_SYSLOG  = cfg.getboolean( 'DEFAULT', 'USE_SYSLOG' )
508
509    except ConfigParser.NoOptionError:
510
511        USE_SYSLOG  = True
512
513        debug_msg( 0, 'ERROR: no option USE_SYSLOG found: assuming yes' )
514
515    if USE_SYSLOG:
516
517        try:
518            SYSLOG_LEVEL = cfg.getint( 'DEFAULT', 'SYSLOG_LEVEL' )
519
520        except ConfigParser.NoOptionError:
521
522            debug_msg( 0, 'ERROR: no option SYSLOG_LEVEL found: assuming level 0' )
523            SYSLOG_LEVEL = 0
524
525        try:
526
527            SYSLOG_FACILITY = eval( 'syslog.LOG_' + cfg.get( 'DEFAULT', 'SYSLOG_FACILITY' ) )
528
529        except ConfigParser.NoOptionError:
530
531            SYSLOG_FACILITY = syslog.LOG_DAEMON
532
533            debug_msg( 0, 'ERROR: no option SYSLOG_FACILITY found: assuming facility DAEMON' )
534
535    try:
536
537        BATCH_SERVER = cfg.get( 'DEFAULT', 'BATCH_SERVER' )
538
539    except ConfigParser.NoOptionError:
540
541        # Not required for all API's: only pbs api allows remote connections
542        BATCH_SERVER = None
543
544    try:
545   
546        BATCH_POLL_INTERVAL = cfg.getint( 'DEFAULT', 'BATCH_POLL_INTERVAL' )
547
548    except ConfigParser.NoOptionError:
549
550        # Backwards compatibility for old configs
551        #
552
553        BATCH_POLL_INTERVAL = cfg.getint( 'DEFAULT', 'TORQUE_POLL_INTERVAL' )
554        api_guess           = 'pbs'
555   
556    try:
557
558        GMOND_CONF          = cfg.get( 'DEFAULT', 'GMOND_CONF' )
559
560    except ConfigParser.NoOptionError:
561
562        # Not specified: assume /etc/ganglia/gmond.conf
563        #
564        GMOND_CONF          = '/etc/ganglia/gmond.conf'
565
566    ganglia_cfg             = GangliaConfigParser( GMOND_CONF )
567    GMETRIC_TARGET          = None
568
569    GMOND_UDP_SEND_CHANNELS = ganglia_cfg.getUdpSendChannels()
570
571    if not GMOND_UDP_SEND_CHANNELS:
572
573        debug_msg( 0, "WARNING: Can't parse udp_send_channel from: '%s' - Trying: %s" %( GMOND_CONF, JOBMOND_CONF ) )
574
575        # Couldn't figure it out: let's see if it's in our jobmond.conf
576        #
577        try:
578
579            GMETRIC_TARGET    = cfg.get( 'DEFAULT', 'GMETRIC_TARGET' )
580
581        # Guess not: now just give up
582       
583        except ConfigParser.NoOptionError:
584
585            GMETRIC_TARGET    = None
586
587            debug_msg( 0, "ERROR: GMETRIC_TARGET not set: internal Gmetric handling aborted. Failing back to DEPRECATED use of gmond.conf/gmetric binary. This will slow down jobmond significantly!" )
588
589            gmetric_bin    = findGmetric()
590
591            if gmetric_bin:
592
593                GMETRIC_BINARY     = gmetric_bin
594            else:
595                debug_msg( 0, "WARNING: Can't find gmetric binary anywhere in $PATH" )
596
597                try:
598
599                    GMETRIC_BINARY = cfg.get( 'DEFAULT', 'GMETRIC_BINARY' )
600
601                except ConfigParser.NoOptionError:
602
603                    print "FATAL ERROR: GMETRIC_BINARY not set and not in $PATH"
604                    sys.exit( 1 )
605
606    #TODO: is this really still needed or should be automatic
607    DETECT_TIME_DIFFS    = cfg.getboolean( 'DEFAULT', 'DETECT_TIME_DIFFS' )
608
609    BATCH_HOST_TRANSLATE = getlist( cfg.get( 'DEFAULT', 'BATCH_HOST_TRANSLATE' ) )
610
611    try:
612
613        BATCH_API    = cfg.get( 'DEFAULT', 'BATCH_API' )
614
615    except ConfigParser.NoOptionError, detail:
616
617        print "FATAL ERROR: BATCH_API not set"
618        sys.exit( 1 )
619
620    try:
621
622        QUEUE        = getlist( cfg.get( 'DEFAULT', 'QUEUE' ) )
623
624    except ConfigParser.NoOptionError, detail:
625
626        QUEUE        = None
627
628    METRIC_MAX_VAL_LEN = ganglia_cfg.getInt( 'globals', 'max_udp_msg_len' )
629
630    return True
631
632def fqdn_parts (fqdn):
633
634    """Return pair of host and domain for fully-qualified domain name arg."""
635
636    parts = fqdn.split (".")
637
638    return (parts[0], string.join(parts[1:], "."))
639
640class DataProcessor:
641
642    """Class for processing of data"""
643
644    binary = None
645
646    def __init__( self, binary=None ):
647
648        """Remember alternate binary location if supplied"""
649
650        global GMETRIC_BINARY, GMOND_CONF
651
652        if binary:
653            self.binary = binary
654
655        if not self.binary and not GMETRIC_TARGET and not GMOND_UDP_SEND_CHANNELS:
656            self.binary = GMETRIC_BINARY
657
658        # Timeout for XML
659        #
660        # From ganglia's documentation:
661        #
662        # 'A metric will be deleted DMAX seconds after it is received, and
663        # DMAX=0 means eternal life.'
664
665        self.dmax = str( int( int( BATCH_POLL_INTERVAL ) * 2 ) )
666
667        if GMOND_CONF and not GMETRIC_TARGET and not GMOND_UDP_SEND_CHANNELS:
668
669            incompatible = self.checkGmetricVersion()
670
671            if incompatible:
672
673                print 'Ganglia/Gmetric version not compatible, please upgrade to at least 3.3.8'
674                sys.exit( 1 )
675
676    def checkGmetricVersion( self ):
677
678        """
679        Check version of gmetric is at least 3.3.8
680        for the syntax we use
681        """
682
683        global METRIC_MAX_VAL_LEN, GMETRIC_TARGET
684
685        incompatible    = 0
686
687        gfp        = os.popen( self.binary + ' --version' )
688        lines      = gfp.readlines()
689
690        gfp.close()
691
692        for line in lines:
693
694            line = line.split( ' ' )
695
696            if len( line ) == 2 and str( line ).find( 'gmetric' ) != -1:
697           
698                gmetric_version    = line[1].split( '\n' )[0]
699
700                version_major    = int( gmetric_version.split( '.' )[0] )
701                version_minor    = int( gmetric_version.split( '.' )[1] )
702                version_patch    = int( gmetric_version.split( '.' )[2] )
703
704                incompatible    = 0
705
706                if version_major < 3:
707
708                    incompatible = 1
709               
710                elif version_major == 3:
711
712                    if version_minor < 3:
713
714                        incompatible = 1
715
716                    elif version_patch < 8:
717
718                        incompatible = 1
719
720        return incompatible
721
722    def multicastGmetric( self, metricname, metricval, valtype='string', units='' ):
723
724        """Call gmetric binary and multicast"""
725
726        cmd = self.binary
727
728        if GMOND_UDP_SEND_CHANNELS:
729
730            for c_ip, c_port  in GMOND_UDP_SEND_CHANNELS:
731
732                metric_debug        = "[gmetric %s:%s] name: %s - val: %s - dmax: %s" %( str(c_ip), str(c_port), str( metricname ), str( metricval ), str( self.dmax ) )
733
734                debug_msg( 10, printTime() + ' ' + metric_debug)
735
736                gm = Gmetric( c_ip, c_port )
737
738                gm.send( str( metricname ), str( metricval ), str( self.dmax ), valtype, units )
739
740        elif GMETRIC_TARGET:
741
742            GMETRIC_TARGET_HOST    = GMETRIC_TARGET.split( ':' )[0]
743            GMETRIC_TARGET_PORT    = GMETRIC_TARGET.split( ':' )[1]
744
745            metric_debug        = "[gmetric] name: %s - val: %s - dmax: %s" %( str( metricname ), str( metricval ), str( self.dmax ) )
746
747            debug_msg( 10, printTime() + ' ' + metric_debug)
748
749            gm = Gmetric( GMETRIC_TARGET_HOST, GMETRIC_TARGET_PORT )
750
751            gm.send( str( metricname ), str( metricval ), str( self.dmax ), valtype, units )
752
753        else:
754            try:
755                cmd = cmd + ' -c' + GMOND_CONF
756
757            except NameError:
758
759                debug_msg( 10, 'Assuming /etc/ganglia/gmond.conf for gmetric cmd' )
760
761            cmd = cmd + ' -n' + str( metricname )+ ' -v"' + str( metricval )+ '" -t' + str( valtype ) + ' -d' + str( self.dmax )
762
763            if len( units ) > 0:
764
765                cmd = cmd + ' -u"' + units + '"'
766
767            debug_msg( 10, printTime() + ' ' + cmd )
768
769            os.system( cmd )
770
771class DataGatherer:
772
773    """Skeleton class for batch system DataGatherer"""
774
775    def printJobs( self, jobs ):
776
777        """Print a jobinfo overview"""
778
779        for name, attrs in self.jobs.items():
780
781            print 'job %s' %(name)
782
783            for name, val in attrs.items():
784
785                print '\t%s = %s' %( name, val )
786
787    def printJob( self, jobs, job_id ):
788
789        """Print job with job_id from jobs"""
790
791        print 'job %s' %(job_id)
792
793        for name, val in jobs[ job_id ].items():
794
795            print '\t%s = %s' %( name, val )
796
797    def getAttr( self, attrs, name ):
798
799        """Return certain attribute from dictionary, if exists"""
800
801        if attrs.has_key( name ):
802
803            return attrs[ name ]
804        else:
805            return ''
806
807    def jobDataChanged( self, jobs, job_id, attrs ):
808
809        """Check if job with attrs and job_id in jobs has changed"""
810
811        if jobs.has_key( job_id ):
812
813            oldData = jobs[ job_id ]   
814        else:
815            return 1
816
817        for name, val in attrs.items():
818
819            if oldData.has_key( name ):
820
821                if oldData[ name ] != attrs[ name ]:
822
823                    return 1
824
825            else:
826                return 1
827
828        return 0
829
830    def submitJobData( self ):
831
832        """Submit job info list"""
833
834        global BATCH_API
835
836        self.dp.multicastGmetric( 'zplugin_monarch_heartbeat', str( int( int( self.cur_time ) + int( self.timeoffset ) ) ) )
837
838        running_jobs = 0
839        queued_jobs  = 0
840
841        # Count how many running/queued jobs we found
842        #
843        for jobid, jobattrs in self.jobs.items():
844
845            if jobattrs[ 'status' ] == 'Q':
846
847                queued_jobs += 1
848
849            elif jobattrs[ 'status' ] == 'R':
850
851                running_jobs += 1
852
853        # Report running/queued jobs as seperate metric for a nice RRD graph
854        #
855        self.dp.multicastGmetric( 'zplugin_monarch_rj', str( running_jobs ), 'uint32', 'jobs' )
856        self.dp.multicastGmetric( 'zplugin_monarch_qj', str( queued_jobs ), 'uint32', 'jobs' )
857
858        # Report down/offline nodes in batch (PBS only ATM)
859        #
860        if BATCH_API in [ 'pbs', 'slurm' ]:
861
862            domain        = fqdn_parts( socket.getfqdn() )[1]
863
864            downed_nodes  = list()
865            offline_nodes = list()
866       
867            l        = ['state']
868
869            nodelist = self.getNodeData()
870
871            for name, node in nodelist.items():
872
873                if ( node[ 'state' ].find( "down" ) != -1 ):
874
875                    downed_nodes.append( name )
876
877                if ( node[ 'state' ].find( "offline" ) != -1 ):
878
879                    offline_nodes.append( name )
880
881            downnodeslist    = do_nodelist( downed_nodes )
882            offlinenodeslist = do_nodelist( offline_nodes )
883
884            down_str    = 'nodes=%s domain=%s reported=%s' %( string.join( downnodeslist, ';' ), domain, str( int( int( self.cur_time ) + int( self.timeoffset ) ) ) )
885            offl_str    = 'nodes=%s domain=%s reported=%s' %( string.join( offlinenodeslist, ';' ), domain, str( int( int( self.cur_time ) + int( self.timeoffset ) ) ) )
886            self.dp.multicastGmetric( 'zplugin_monarch_down'   , down_str )
887            self.dp.multicastGmetric( 'zplugin_monarch_offline', offl_str )
888
889        # Now let's spread the knowledge
890        #
891        for jobid, jobattrs in self.jobs.items():
892
893            # Make gmetric values for each job: respect max gmetric value length
894            #
895            gmetric_val        = self.compileGmetricVal( jobid, jobattrs )
896            metric_increment    = 0
897
898            # If we have more job info than max gmetric value length allows, split it up
899            # amongst multiple metrics
900            #
901            for val in gmetric_val:
902
903                metric_name = 'zplugin_monarch_job_%s_%s' %( str(metric_increment) , str( jobid ) )
904                self.dp.multicastGmetric( metric_name, val )
905
906                # Increase follow number if this jobinfo is split up amongst more than 1 gmetric
907                #
908                metric_increment    = metric_increment + 1
909
910    def compileGmetricVal( self, jobid, jobattrs ):
911
912        """Create a val string for gmetric of jobinfo"""
913
914        gval_lists    = [ ]
915        val_list    = { }
916
917        for val_name, val_value in jobattrs.items():
918
919            # These are our own metric names, i.e.: status, start_timestamp, etc
920            #
921            val_list_names_len    = len( string.join( val_list.keys() ) ) + len(val_list.keys())
922
923            # These are their corresponding values
924            #
925            val_list_vals_len    = len( string.join( val_list.values() ) ) + len(val_list.values())
926
927            if val_name == 'nodes' and jobattrs['status'] == 'R':
928
929                node_str = None
930
931                for node in val_value:
932
933                    if node_str:
934
935                        node_str = node_str + ';' + node
936                    else:
937                        node_str = node
938
939                    # Make sure if we add this new info, that the total metric's value length does not exceed METRIC_MAX_VAL_LEN
940                    #
941                    if (val_list_names_len + len(val_name) ) + (val_list_vals_len + len(node_str) ) > METRIC_MAX_VAL_LEN:
942
943                        # It's too big, we need to make a new gmetric for the additional info
944                        #
945                        val_list[ val_name ]    = node_str
946
947                        gval_lists.append( val_list )
948
949                        val_list        = { }
950                        node_str        = None
951
952                val_list[ val_name ]    = node_str
953
954                gval_lists.append( val_list )
955
956                val_list        = { }
957
958            elif val_value != '':
959
960                # Make sure if we add this new info, that the total metric's value length does not exceed METRIC_MAX_VAL_LEN
961                #
962                if (val_list_names_len + len(val_name) ) + (val_list_vals_len + len(str(val_value)) ) > METRIC_MAX_VAL_LEN:
963
964                    # It's too big, we need to make a new gmetric for the additional info
965                    #
966                    gval_lists.append( val_list )
967
968                    val_list        = { }
969
970                val_list[ val_name ]    = val_value
971
972        if len( val_list ) > 0:
973
974            gval_lists.append( val_list )
975
976        str_list    = [ ]
977
978        # Now append the value names and values together, i.e.: stop_timestamp=value, etc
979        #
980        for val_list in gval_lists:
981
982            my_val_str    = None
983
984            for val_name, val_value in val_list.items():
985
986                if type(val_value) == list:
987
988                    val_value    = val_value.join( ',' )
989
990                if my_val_str:
991
992                    try:
993                        # fixme: It's getting
994                        # ('nodes', None) items
995                        my_val_str = my_val_str + ' ' + val_name + '=' + val_value
996                    except:
997                        pass
998
999                else:
1000                    my_val_str = val_name + '=' + val_value
1001
1002            str_list.append( my_val_str )
1003
1004        return str_list
1005
1006    def daemon( self ):
1007
1008        """Run as daemon forever"""
1009
1010        # Fork the first child
1011        #
1012        pid = os.fork()
1013        if pid > 0:
1014            sys.exit(0)  # end parent
1015
1016        # creates a session and sets the process group ID
1017        #
1018        os.setsid()
1019
1020        # Fork the second child
1021        #
1022        pid = os.fork()
1023        if pid > 0:
1024            sys.exit(0)  # end parent
1025
1026        write_pidfile()
1027
1028        # Go to the root directory and set the umask
1029        #
1030        os.chdir('/')
1031        os.umask(0)
1032
1033        sys.stdin.close()
1034        sys.stdout.close()
1035        sys.stderr.close()
1036
1037        os.open('/dev/null', os.O_RDWR)
1038        os.dup2(0, 1)
1039        os.dup2(0, 2)
1040
1041        self.run()
1042
1043    def run( self ):
1044
1045        """Main thread"""
1046
1047        while ( 1 ):
1048       
1049            self.getJobData()
1050            self.submitJobData()
1051            time.sleep( BATCH_POLL_INTERVAL )   
1052
1053# SGE code by Dave Love <fx@gnu.org>.  Tested with SGE 6.0u8 and 6.0u11.  May
1054# work with SGE 6.1 (else should be easily fixable), but definitely doesn't
1055# with 6.2.  See also the fixmes.
1056
1057class NoJobs (Exception):
1058    """Exception raised by empty job list in qstat output."""
1059    pass
1060
1061class SgeQstatXMLParser(xml.sax.handler.ContentHandler):
1062    """SAX handler for XML output from Sun Grid Engine's `qstat'."""
1063
1064    def __init__(self):
1065        self.value = ""
1066        self.joblist = []
1067        self.job = {}
1068        self.queue = ""
1069        self.in_joblist = False
1070        self.lrequest = False
1071        self.eltq = deque()
1072        xml.sax.handler.ContentHandler.__init__(self)
1073
1074    # The structure of the output is as follows (for SGE 6.0).  It's
1075    # similar for 6.1, but radically different for 6.2, and is
1076    # undocumented generally.  Unfortunately it's voluminous, and probably
1077    # doesn't scale to large clusters/queues.
1078
1079    # <detailed_job_info  xmlns:xsd="http://www.w3.org/2001/XMLSchema">
1080    #   <djob_info>
1081    #     <qmaster_response>  <!-- job -->
1082    #       ...
1083    #       <JB_ja_template> 
1084    #     <ulong_sublist>
1085    #     ...         <!-- start_time, state ... -->
1086    #     </ulong_sublist>
1087    #       </JB_ja_template> 
1088    #       <JB_ja_tasks>
1089    #     <ulong_sublist>
1090    #       ...       <!-- task info
1091    #     </ulong_sublist>
1092    #     ...
1093    #       </JB_ja_tasks>
1094    #       ...
1095    #     </qmaster_response>
1096    #   </djob_info>
1097    #   <messages>
1098    #   ...
1099
1100    # NB.  We might treat each task as a separate job, like
1101    # straight qstat output, but the web interface expects jobs to
1102    # be identified by integers, not, say, <job number>.<task>.
1103
1104    # So, I lied.  If the job list is empty, we get invalid XML
1105    # like this, which we need to defend against:
1106
1107    # <unknown_jobs  xmlns:xsd="http://www.w3.org/2001/XMLSchema">
1108    #   <>
1109    #     <ST_name>*</ST_name>
1110    #   </>
1111    # </unknown_jobs>
1112
1113    def startElement(self, name, attrs):
1114        self.value = ""
1115        if name == "djob_info":    # job list
1116            self.in_joblist = True
1117        # The job container is "qmaster_response" in SGE 6.0
1118        # and 6.1, but "element" in 6.2.  This is only the very
1119        # start of what's necessary for 6.2, though (sigh).
1120        elif (name == "qmaster_response" or name == "element") \
1121                and self.eltq[-1] == "djob_info": # job
1122            self.job = {"job_state": "U", "slots": 0,
1123                    "nodes": [], "queued_timestamp": "",
1124                    "queued_timestamp": "", "queue": "",
1125                    "ppn": "0", "RN_max": 0,
1126                    # fixme in endElement
1127                    "requested_memory": 0, "requested_time": 0
1128                    }
1129            self.joblist.append(self.job)
1130        elif name == "qstat_l_requests": # resource request
1131            self.lrequest = True
1132        elif name == "unknown_jobs":
1133            raise NoJobs
1134        self.eltq.append (name)
1135
1136    def characters(self, ch):
1137        self.value += ch
1138
1139    def endElement(self, name): 
1140        """Snarf job elements contents into job dictionary.
1141           Translate keys if appropriate."""
1142
1143        name_trans = {
1144          "JB_job_number": "number",
1145          "JB_job_name": "name", "JB_owner": "owner",
1146          "queue_name": "queue", "JAT_start_time": "start_timestamp",
1147          "JB_submission_time": "queued_timestamp"
1148          }
1149        value = self.value
1150        self.eltq.pop ()
1151
1152        if name == "djob_info":
1153            self.in_joblist = False
1154            self.job = {}
1155        elif name == "JAT_master_queue":
1156            self.job["queue"] = value.split("@")[0]
1157        elif name == "JG_qhostname":
1158            if not (value in self.job["nodes"]):
1159                self.job["nodes"].append(value)
1160        elif name == "JG_slots": # slots in use
1161            self.job["slots"] += int(value)
1162        elif name == "RN_max": # requested slots (tasks or parallel)
1163            self.job["RN_max"] = max (self.job["RN_max"],
1164                          int(value))
1165        elif name == "JAT_state": # job state (bitwise or)
1166            value = int (value)
1167            # Status values from sge_jobL.h
1168            #define JIDLE           0x00000000
1169            #define JHELD           0x00000010
1170            #define JMIGRATING          0x00000020
1171            #define JQUEUED         0x00000040
1172            #define JRUNNING        0x00000080
1173            #define JSUSPENDED          0x00000100
1174            #define JTRANSFERING        0x00000200
1175            #define JDELETED        0x00000400
1176            #define JWAITING        0x00000800
1177            #define JEXITING        0x00001000
1178            #define JWRITTEN        0x00002000
1179            #define JSUSPENDED_ON_THRESHOLD 0x00010000
1180            #define JFINISHED           0x00010000
1181            if value & 0x80:
1182                self.job["status"] = "R"
1183            elif value & 0x40:
1184                self.job["status"] = "Q"
1185            else:
1186                self.job["status"] = "O" # `other'
1187        elif name == "CE_name" and self.lrequest and self.value in \
1188                ("h_cpu", "s_cpu", "cpu", "h_core", "s_core"):
1189            # We're in a container for an interesting resource
1190            # request; record which type.
1191            self.lrequest = self.value
1192        elif name == "CE_doubleval" and self.lrequest:
1193            # if we're in a container for an interesting
1194            # resource request, use the maxmimum of the hard
1195            # and soft requests to record the requested CPU
1196            # or core.  Fixme:  I'm not sure if this logic is
1197            # right.
1198            if self.lrequest in ("h_core", "s_core"):
1199                self.job["requested_memory"] = \
1200                    max (float (value),
1201                     self.job["requested_memory"])
1202            # Fixme:  Check what cpu means, c.f [hs]_cpu.
1203            elif self.lrequest in ("h_cpu", "s_cpu", "cpu"):
1204                self.job["requested_time"] = \
1205                    max (float (value),
1206                     self.job["requested_time"])
1207        elif name == "qstat_l_requests":
1208            self.lrequest = False
1209        elif self.job and self.in_joblist:
1210            if name in name_trans:
1211                name = name_trans[name]
1212                self.job[name] = value
1213
1214# Abstracted from PBS original.
1215# Fixme:  Is it worth (or appropriate for PBS) sorting the result?
1216#
1217def do_nodelist( nodes ):
1218
1219    """Translate node list as appropriate."""
1220
1221    nodeslist        = [ ]
1222    my_domain        = fqdn_parts( socket.getfqdn() )[1]
1223
1224    for node in nodes:
1225
1226        host        = node.split( '/' )[0] # not relevant for SGE
1227        h, host_domain    = fqdn_parts(host)
1228
1229        if host_domain == my_domain:
1230
1231            host    = h
1232
1233        if nodeslist.count( host ) == 0:
1234
1235            for translate_pattern in BATCH_HOST_TRANSLATE:
1236
1237                if translate_pattern.find( '/' ) != -1:
1238
1239                    translate_orig    = \
1240                        translate_pattern.split( '/' )[1]
1241                    translate_new    = \
1242                        translate_pattern.split( '/' )[2]
1243                    host = re.sub( translate_orig,
1244                               translate_new, host )
1245            if not host in nodeslist:
1246                nodeslist.append( host )
1247    return nodeslist
1248
1249class SLURMDataGatherer( DataGatherer ):
1250
1251    global pyslurm
1252
1253    """This is the DataGatherer for SLURM"""
1254
1255    def __init__( self ):
1256
1257        """Setup appropriate variables"""
1258
1259        self.jobs       = { }
1260        self.timeoffset = 0
1261        self.dp         = DataProcessor()
1262
1263    def getNodeData( self ):
1264
1265        slurm_type  = pyslurm.node()
1266
1267        slurm_nodes = slurm_type.get()
1268
1269        nodedict    = { }
1270
1271        for node, attrs in slurm_nodes.items():
1272
1273            ( num_state, name_state ) = attrs['node_state'] 
1274
1275            if name_state == 'DOWN':
1276
1277                nodedict[ node ] = { 'state' : 'down' }
1278
1279            elif name_state == 'DRAIN':
1280
1281                nodedict[ node ] = { 'state' : 'offline' }
1282
1283        return nodedict
1284
1285    def getJobData( self ):
1286
1287        """Gather all data on current jobs"""
1288
1289        joblist            = {}
1290
1291        self.cur_time  = time.time()
1292
1293        slurm_type = pyslurm.job()
1294        joblist    = slurm_type.get()
1295
1296        jobs_processed    = [ ]
1297
1298        for name, attrs in joblist.items():
1299            display_queue = 1
1300            job_id        = name
1301
1302            name          = self.getAttr( attrs, 'name' )
1303            queue         = self.getAttr( attrs, 'partition' )
1304
1305            if QUEUE:
1306                for q in QUEUE:
1307                    if q == queue:
1308                        display_queue = 1
1309                        break
1310                    else:
1311                        display_queue = 0
1312                        continue
1313            if display_queue == 0:
1314                continue
1315
1316            owner_uid        = attrs[ 'user_id' ]
1317            ( owner, owner_pw, owner_uid, owner_gid, owner_gecos, owner_dir, owner_shell ) = pwd.getpwuid( owner_uid )
1318
1319            requested_time   = self.getAttr( attrs, 'time_limit' )
1320            min_memory       = self.getAttr( attrs, 'pn_min_memory' )
1321
1322            if min_memory == 0:
1323
1324                requested_memory = ''
1325
1326            else:
1327                requested_memory = min_memory
1328
1329            min_cpus = self.getAttr( attrs, 'pn_min_cpus' )
1330
1331            if min_cpus == 0:
1332
1333                ppn = ''
1334
1335            else:
1336                ppn = min_cpus
1337
1338            ( something, status_long ) = self.getAttr( attrs, 'job_state' )
1339
1340            status = 'Q'
1341
1342            if status_long == 'RUNNING':
1343
1344                status = 'R'
1345
1346            elif status_long == 'COMPLETED':
1347
1348                continue
1349
1350            jobs_processed.append( job_id )
1351
1352            queued_timestamp = self.getAttr( attrs, 'submit_time' )
1353
1354            start_timestamp = ''
1355            nodeslist       = ''
1356
1357            if status == 'R':
1358
1359                start_timestamp = self.getAttr( attrs, 'start_time' )
1360                nodes           = attrs[ 'nodes' ]
1361
1362                if not nodes:
1363
1364                    # This should not happen
1365
1366                    # Something wrong: running but 'nodes' returned empty by pyslurm
1367                    # Possible pyslurm bug: abort/quit/warning
1368
1369                    err_msg = 'FATAL ERROR: job %s running but nodes returned empty: pyslurm bugged?' %job_id
1370
1371                    print err_msg
1372                    debug_msg( 0, err_msg )
1373                    sys.exit(1)
1374
1375                my_nodelist = [ ]
1376
1377                slurm_hostlist  = pyslurm.hostlist()
1378                slurm_hostlist.create( nodes )
1379                slurm_hostlist.uniq()
1380
1381                while slurm_hostlist.count() > 0:
1382
1383                    my_nodelist.append( slurm_hostlist.pop() )
1384
1385                slurm_hostlist.destroy()
1386
1387                del slurm_hostlist
1388
1389                nodeslist       = do_nodelist( my_nodelist )
1390
1391                if DETECT_TIME_DIFFS:
1392
1393                    # If a job start if later than our current date,
1394                    # that must mean the Torque server's time is later
1395                    # than our local time.
1396               
1397                    if int( start_timestamp ) > int( int( self.cur_time ) + int( self.timeoffset ) ):
1398
1399                        self.timeoffset = int( int(start_timestamp) - int(self.cur_time) )
1400
1401            elif status == 'Q':
1402
1403                nodeslist       = str( attrs[ 'num_nodes' ] )
1404
1405            else:
1406                start_timestamp = ''
1407                nodeslist       = ''
1408
1409            myAttrs                = { }
1410
1411            myAttrs[ 'name' ]             = str( name )
1412            myAttrs[ 'queue' ]            = str( queue )
1413            myAttrs[ 'owner' ]            = str( owner )
1414            myAttrs[ 'requested_time' ]   = str( requested_time )
1415            myAttrs[ 'requested_memory' ] = str( requested_memory )
1416            myAttrs[ 'ppn' ]              = str( ppn )
1417            myAttrs[ 'status' ]           = str( status )
1418            myAttrs[ 'start_timestamp' ]  = str( start_timestamp )
1419            myAttrs[ 'queued_timestamp' ] = str( queued_timestamp )
1420            myAttrs[ 'reported' ]         = str( int( int( self.cur_time ) + int( self.timeoffset ) ) )
1421            myAttrs[ 'nodes' ]            = nodeslist
1422            myAttrs[ 'domain' ]           = fqdn_parts( socket.getfqdn() )[1]
1423            myAttrs[ 'poll_interval' ]    = str( BATCH_POLL_INTERVAL )
1424
1425            if self.jobDataChanged( self.jobs, job_id, myAttrs ) and myAttrs['status'] in [ 'R', 'Q' ]:
1426
1427                self.jobs[ job_id ] = myAttrs
1428
1429        for id, attrs in self.jobs.items():
1430
1431            if id not in jobs_processed:
1432
1433                # This one isn't there anymore; toedeledoki!
1434                #
1435                del self.jobs[ id ]
1436
1437class SgeDataGatherer(DataGatherer):
1438
1439    jobs = {}
1440
1441    def __init__( self ):
1442        self.jobs = {}
1443        self.timeoffset = 0
1444        self.dp = DataProcessor()
1445
1446    def getJobData( self ):
1447        """Gather all data on current jobs in SGE"""
1448
1449        import popen2
1450
1451        self.cur_time = 0
1452        queues = ""
1453        if QUEUE:    # only for specific queues
1454            # Fixme:  assumes queue names don't contain single
1455            # quote or comma.  Don't know what the SGE rules are.
1456            queues = " -q '" + string.join (QUEUE, ",") + "'"
1457        # Note the comment in SgeQstatXMLParser about scaling with
1458        # this method of getting data.  I haven't found better one.
1459        # Output with args `-xml -ext -f -r' is easier to parse
1460        # in some ways, harder in others, but it doesn't provide
1461        # the submission time (at least SGE 6.0).  The pipeline
1462        # into sed corrects bogus XML observed with a configuration
1463        # of SGE 6.0u8, which otherwise causes the parsing to hang.
1464        piping = popen2.Popen3("qstat -u '*' -j '*' -xml | \
1465sed -e 's/reported usage>/reported_usage>/g' -e 's;<\/*JATASK:.*>;;'" \
1466                           + queues, True)
1467        qstatparser = SgeQstatXMLParser()
1468        parse_err = 0
1469        try:
1470            xml.sax.parse(piping.fromchild, qstatparser)
1471        except NoJobs:
1472            pass
1473        except:
1474            parse_err = 1
1475        if piping.wait():
1476            debug_msg(10, "qstat error, skipping until next polling interval: " + piping.childerr.readline())
1477            return None
1478        elif parse_err:
1479            debug_msg(10, "Bad XML output from qstat"())
1480            exit (1)
1481        for f in piping.fromchild, piping.tochild, piping.childerr:
1482            f.close()
1483        self.cur_time = time.time()
1484        jobs_processed = []
1485        for job in qstatparser.joblist:
1486            job_id = job["number"]
1487            if job["status"] in [ 'Q', 'R' ]:
1488                jobs_processed.append(job_id)
1489            if job["status"] == "R":
1490                job["nodes"] = do_nodelist (job["nodes"])
1491                # Fixme: why is job["nodes"] sometimes null?
1492                try:
1493                    # Fixme: Is this sensible?  The
1494                    # PBS-type PPN isn't something you use
1495                    # with SGE.
1496                    job["ppn"] = float(job["slots"]) / len(job["nodes"])
1497                except:
1498                    job["ppn"] = 0
1499                if DETECT_TIME_DIFFS:
1500                    # If a job start is later than our
1501                    # current date, that must mean
1502                    # the SGE server's time is later
1503                    # than our local time.
1504                    start_timestamp = int (job["start_timestamp"])
1505                    if start_timestamp > int(self.cur_time) + int(self.timeoffset):
1506
1507                        self.timeoffset    = start_timestamp - int(self.cur_time)
1508            else:
1509                # fixme: Note sure what this should be:
1510                job["ppn"] = job["RN_max"]
1511                job["nodes"] = "1"
1512
1513            myAttrs = {}
1514            for attr in ["name", "queue", "owner",
1515                     "requested_time", "status",
1516                     "requested_memory", "ppn",
1517                     "start_timestamp", "queued_timestamp"]:
1518                myAttrs[attr] = str(job[attr])
1519            myAttrs["nodes"] = job["nodes"]
1520            myAttrs["reported"] = str(int(self.cur_time) + int(self.timeoffset))
1521            myAttrs["domain"] = fqdn_parts(socket.getfqdn())[1]
1522            myAttrs["poll_interval"] = str(BATCH_POLL_INTERVAL)
1523
1524            if self.jobDataChanged(self.jobs, job_id, myAttrs) and myAttrs["status"] in ["R", "Q"]:
1525                self.jobs[job_id] = myAttrs
1526        for id, attrs in self.jobs.items():
1527            if id not in jobs_processed:
1528                del self.jobs[id]
1529
1530# LSF code by Mahmoud Hanafi <hanafim@users.sourceforge.nt>
1531# Requres LSFObject http://sourceforge.net/projects/lsfobject
1532#
1533class LsfDataGatherer(DataGatherer):
1534
1535    """This is the DataGatherer for LSf"""
1536
1537    global lsfObject
1538
1539    def __init__( self ):
1540
1541        self.jobs = { }
1542        self.timeoffset = 0
1543        self.dp = DataProcessor()
1544        self.initLsfQuery()
1545
1546    def _countDuplicatesInList( self, dupedList ):
1547
1548        countDupes    = { }
1549
1550        for item in dupedList:
1551
1552            if not countDupes.has_key( item ):
1553
1554                countDupes[ item ]    = 1
1555            else:
1556                countDupes[ item ]    = countDupes[ item ] + 1
1557
1558        dupeCountList    = [ ]
1559
1560        for item, count in countDupes.items():
1561
1562            dupeCountList.append( ( item, count ) )
1563
1564        return dupeCountList
1565#
1566#lst = ['I1','I2','I1','I3','I4','I4','I7','I7','I7','I7','I7']
1567#print _countDuplicatesInList(lst)
1568#[('I1', 2), ('I3', 1), ('I2', 1), ('I4', 2), ('I7', 5)]
1569########################
1570
1571    def initLsfQuery( self ):
1572        self.pq = None
1573        self.pq = lsfObject.jobInfoEntObject()
1574
1575    def getJobData( self, known_jobs="" ):
1576        """Gather all data on current jobs in LSF"""
1577        if len( known_jobs ) > 0:
1578            jobs = known_jobs
1579        else:
1580            jobs = { }
1581        joblist = {}
1582        joblist = self.pq.getJobInfo()
1583        nodelist = ''
1584
1585        self.cur_time = time.time()
1586
1587        jobs_processed = [ ]
1588
1589        for name, attrs in joblist.items():
1590            job_id = str(name)
1591            jobs_processed.append( job_id )
1592            name = self.getAttr( attrs, 'jobName' )
1593            queue = self.getAttr( self.getAttr( attrs, 'submit') , 'queue' )
1594            owner = self.getAttr( attrs, 'user' )
1595
1596### THIS IS THE rLimit List index values
1597#define LSF_RLIMIT_CPU      0        /* cpu time in milliseconds */
1598#define LSF_RLIMIT_FSIZE    1        /* maximum file size */
1599#define LSF_RLIMIT_DATA     2        /* data size */
1600#define LSF_RLIMIT_STACK    3        /* stack size */
1601#define LSF_RLIMIT_CORE     4        /* core file size */
1602#define LSF_RLIMIT_RSS      5        /* resident set size */
1603#define LSF_RLIMIT_NOFILE   6        /* open files */
1604#define LSF_RLIMIT_OPEN_MAX 7        /* (from HP-UX) */
1605#define LSF_RLIMIT_VMEM     8        /* maximum swap mem */
1606#define LSF_RLIMIT_SWAP     8
1607#define LSF_RLIMIT_RUN      9        /* max wall-clock time limit */
1608#define LSF_RLIMIT_PROCESS  10       /* process number limit */
1609#define LSF_RLIMIT_THREAD   11       /* thread number limit (introduced in LSF6.0) */
1610#define LSF_RLIM_NLIMITS    12       /* number of resource limits */
1611
1612            requested_time = self.getAttr( self.getAttr( attrs, 'submit') , 'rLimits' )[9]
1613            if requested_time == -1: 
1614                requested_time = ""
1615            requested_memory = self.getAttr( self.getAttr( attrs, 'submit') , 'rLimits' )[8]
1616            if requested_memory == -1: 
1617                requested_memory = ""
1618# This tries to get proc per node. We don't support this right now
1619            ppn = 0 #self.getAttr( self.getAttr( attrs, 'SubmitList') , 'numProessors' )
1620            requested_cpus = self.getAttr( self.getAttr( attrs, 'submit') , 'numProcessors' )
1621            if requested_cpus == None or requested_cpus == "":
1622                requested_cpus = 1
1623
1624            if QUEUE:
1625                for q in QUEUE:
1626                    if q == queue:
1627                        display_queue = 1
1628                        break
1629                    else:
1630                        display_queue = 0
1631                        continue
1632            if display_queue == 0:
1633                continue
1634
1635            runState = self.getAttr( attrs, 'status' )
1636            if runState == 4:
1637                status = 'R'
1638            else:
1639                status = 'Q'
1640            queued_timestamp = self.getAttr( attrs, 'submitTime' )
1641
1642            if status == 'R':
1643                start_timestamp = self.getAttr( attrs, 'startTime' )
1644                nodesCpu =  dict(self._countDuplicatesInList(self.getAttr( attrs, 'exHosts' )))
1645                nodelist = nodesCpu.keys()
1646
1647                if DETECT_TIME_DIFFS:
1648
1649                    # If a job start if later than our current date,
1650                    # that must mean the Torque server's time is later
1651                    # than our local time.
1652
1653                    if int(start_timestamp) > int( int(self.cur_time) + int(self.timeoffset) ):
1654
1655                        self.timeoffset = int( int(start_timestamp) - int(self.cur_time) )
1656
1657            elif status == 'Q':
1658                start_timestamp = ''
1659                count_mynodes = 0
1660                numeric_node = 1
1661                nodelist = ''
1662
1663            myAttrs = { }
1664            if name == "":
1665                myAttrs['name'] = "none"
1666            else:
1667                myAttrs['name'] = name
1668
1669            myAttrs[ 'owner' ]        = owner
1670            myAttrs[ 'requested_time' ]    = str(requested_time)
1671            myAttrs[ 'requested_memory' ]    = str(requested_memory)
1672            myAttrs[ 'requested_cpus' ]    = str(requested_cpus)
1673            myAttrs[ 'ppn' ]        = str( ppn )
1674            myAttrs[ 'status' ]        = status
1675            myAttrs[ 'start_timestamp' ]    = str(start_timestamp)
1676            myAttrs[ 'queue' ]        = str(queue)
1677            myAttrs[ 'queued_timestamp' ]    = str(queued_timestamp)
1678            myAttrs[ 'reported' ]        = str( int( int( self.cur_time ) + int( self.timeoffset ) ) )
1679            myAttrs[ 'nodes' ]        = do_nodelist( nodelist )
1680            myAttrs[ 'domain' ]        = fqdn_parts( socket.getfqdn() )[1]
1681            myAttrs[ 'poll_interval' ]    = str(BATCH_POLL_INTERVAL)
1682
1683            if self.jobDataChanged( jobs, job_id, myAttrs ) and myAttrs['status'] in [ 'R', 'Q' ]:
1684                jobs[ job_id ] = myAttrs
1685
1686                debug_msg( 10, printTime() + ' job %s state changed' %(job_id) )
1687
1688        for id, attrs in jobs.items():
1689            if id not in jobs_processed:
1690                # This one isn't there anymore
1691                #
1692                del jobs[ id ]
1693        self.jobs=jobs
1694
1695
1696class PbsDataGatherer( DataGatherer ):
1697
1698    """This is the DataGatherer for PBS and Torque"""
1699
1700    global PBSQuery, PBSError
1701
1702    def __init__( self ):
1703
1704        """Setup appropriate variables"""
1705
1706        self.jobs       = { }
1707        self.timeoffset = 0
1708        self.dp         = DataProcessor()
1709
1710        self.initPbsQuery()
1711
1712    def initPbsQuery( self ):
1713
1714        self.pq = None
1715
1716        try:
1717
1718            if( BATCH_SERVER ):
1719
1720                self.pq = PBSQuery( BATCH_SERVER )
1721            else:
1722                self.pq = PBSQuery()
1723
1724        except PBSError, details:
1725            print 'Cannot connect to pbs server'
1726            print details
1727            sys.exit( 1 )
1728
1729        try:
1730            # TODO: actually use new data structure
1731            self.pq.old_data_structure()
1732
1733        except AttributeError:
1734
1735            # pbs_query is older
1736            #
1737            pass
1738
1739    def getNodeData( self ):
1740
1741        nodedict = { }
1742
1743        try:
1744            nodedict = self.pq.getnodes()
1745
1746        except PBSError, detail:
1747
1748            debug_msg( 10, "PBS server unavailable, skipping until next polling interval: " + str( detail ) )
1749
1750        return nodedict
1751
1752    def getJobData( self ):
1753
1754        """Gather all data on current jobs in Torque"""
1755
1756        joblist            = {}
1757        self.cur_time      = 0
1758
1759        try:
1760            joblist        = self.pq.getjobs()
1761            self.cur_time  = time.time()
1762
1763        except PBSError, detail:
1764
1765            debug_msg( 10, "PBS server unavailable, skipping until next polling interval: " + str( detail ) )
1766            return None
1767
1768        jobs_processed    = [ ]
1769
1770        for name, attrs in joblist.items():
1771            display_queue = 1
1772            job_id        = name.split( '.' )[0]
1773
1774            name          = self.getAttr( attrs, 'Job_Name' )
1775            queue         = self.getAttr( attrs, 'queue' )
1776
1777            if QUEUE:
1778                for q in QUEUE:
1779                    if q == queue:
1780                        display_queue = 1
1781                        break
1782                    else:
1783                        display_queue = 0
1784                        continue
1785            if display_queue == 0:
1786                continue
1787
1788
1789            owner            = self.getAttr( attrs, 'Job_Owner' ).split( '@' )[0]
1790            requested_time   = self.getAttr( attrs, 'Resource_List.walltime' )
1791            requested_memory = self.getAttr( attrs, 'Resource_List.mem' )
1792
1793            mynoderequest    = self.getAttr( attrs, 'Resource_List.nodes' )
1794
1795            ppn = ''
1796
1797            if mynoderequest.find( ':' ) != -1 and mynoderequest.find( 'ppn' ) != -1:
1798
1799                mynoderequest_fields = mynoderequest.split( ':' )
1800
1801                for mynoderequest_field in mynoderequest_fields:
1802
1803                    if mynoderequest_field.find( 'ppn' ) != -1:
1804
1805                        ppn = mynoderequest_field.split( 'ppn=' )[1]
1806
1807            status = self.getAttr( attrs, 'job_state' )
1808
1809            if status in [ 'Q', 'R' ]:
1810
1811                jobs_processed.append( job_id )
1812
1813            queued_timestamp = self.getAttr( attrs, 'ctime' )
1814
1815            if status == 'R':
1816
1817                start_timestamp = self.getAttr( attrs, 'mtime' )
1818                nodes           = self.getAttr( attrs, 'exec_host' ).split( '+' )
1819
1820                nodeslist       = do_nodelist( nodes )
1821
1822                if DETECT_TIME_DIFFS:
1823
1824                    # If a job start if later than our current date,
1825                    # that must mean the Torque server's time is later
1826                    # than our local time.
1827               
1828                    if int( start_timestamp ) > int( int( self.cur_time ) + int( self.timeoffset ) ):
1829
1830                        self.timeoffset = int( int(start_timestamp) - int(self.cur_time) )
1831
1832            elif status == 'Q':
1833
1834                # 'mynodequest' can be a string in the following syntax according to the
1835                # Torque Administator's manual:
1836                #
1837                # {<node_count> | <hostname>}[:ppn=<ppn>][:<property>[:<property>]...][+ ...]
1838                # {<node_count> | <hostname>}[:ppn=<ppn>][:<property>[:<property>]...][+ ...]
1839                # etc
1840                #
1841
1842                #
1843                # For now we only count the amount of nodes request and ignore properties
1844                #
1845
1846                start_timestamp = ''
1847                count_mynodes   = 0
1848
1849                for node in mynoderequest.split( '+' ):
1850
1851                    # Just grab the {node_count|hostname} part and ignore properties
1852                    #
1853                    nodepart     = node.split( ':' )[0]
1854
1855                    # Let's assume a node_count value
1856                    #
1857                    numeric_node = 1
1858
1859                    # Chop the value up into characters
1860                    #
1861                    for letter in nodepart:
1862
1863                        # If this char is not a digit (0-9), this must be a hostname
1864                        #
1865                        if letter not in string.digits:
1866
1867                            numeric_node = 0
1868
1869                    # If this is a hostname, just count this as one (1) node
1870                    #
1871                    if not numeric_node:
1872
1873                        count_mynodes = count_mynodes + 1
1874                    else:
1875
1876                        # If this a number, it must be the node_count
1877                        # and increase our count with it's value
1878                        #
1879                        try:
1880                            count_mynodes = count_mynodes + int( nodepart )
1881
1882                        except ValueError, detail:
1883
1884                            # When we arrive here I must be bugged or very confused
1885                            # THIS SHOULD NOT HAPPEN!
1886                            #
1887                            debug_msg( 10, str( detail ) )
1888                            debug_msg( 10, "Encountered weird node in Resources_List?!" )
1889                            debug_msg( 10, 'nodepart = ' + str( nodepart ) )
1890                            debug_msg( 10, 'job = ' + str( name ) )
1891                            debug_msg( 10, 'attrs = ' + str( attrs ) )
1892                       
1893                nodeslist       = str( count_mynodes )
1894            else:
1895                start_timestamp = ''
1896                nodeslist       = ''
1897
1898            myAttrs                = { }
1899
1900            myAttrs[ 'name' ]             = str( name )
1901            myAttrs[ 'queue' ]            = str( queue )
1902            myAttrs[ 'owner' ]            = str( owner )
1903            myAttrs[ 'requested_time' ]   = str( requested_time )
1904            myAttrs[ 'requested_memory' ] = str( requested_memory )
1905            myAttrs[ 'ppn' ]              = str( ppn )
1906            myAttrs[ 'status' ]           = str( status )
1907            myAttrs[ 'start_timestamp' ]  = str( start_timestamp )
1908            myAttrs[ 'queued_timestamp' ] = str( queued_timestamp )
1909            myAttrs[ 'reported' ]         = str( int( int( self.cur_time ) + int( self.timeoffset ) ) )
1910            myAttrs[ 'nodes' ]            = nodeslist
1911            myAttrs[ 'domain' ]           = fqdn_parts( socket.getfqdn() )[1]
1912            myAttrs[ 'poll_interval' ]    = str( BATCH_POLL_INTERVAL )
1913
1914            if self.jobDataChanged( self.jobs, job_id, myAttrs ) and myAttrs['status'] in [ 'R', 'Q' ]:
1915
1916                self.jobs[ job_id ] = myAttrs
1917
1918        for id, attrs in self.jobs.items():
1919
1920            if id not in jobs_processed:
1921
1922                # This one isn't there anymore; toedeledoki!
1923                #
1924                del self.jobs[ id ]
1925
1926GMETRIC_DEFAULT_TYPE    = 'string'
1927GMETRIC_DEFAULT_HOST    = '127.0.0.1'
1928GMETRIC_DEFAULT_PORT    = '8649'
1929GMETRIC_DEFAULT_UNITS   = ''
1930
1931class Gmetric:
1932
1933    global GMETRIC_DEFAULT_HOST, GMETRIC_DEFAULT_PORT
1934
1935    slope           = { 'zero' : 0, 'positive' : 1, 'negative' : 2, 'both' : 3, 'unspecified' : 4 }
1936    type            = ( '', 'string', 'uint16', 'int16', 'uint32', 'int32', 'float', 'double', 'timestamp' )
1937    protocol        = ( 'udp', 'multicast' )
1938
1939    def __init__( self, host=GMETRIC_DEFAULT_HOST, port=GMETRIC_DEFAULT_PORT ):
1940               
1941        global GMETRIC_DEFAULT_TYPE
1942
1943        self.prot       = self.checkHostProtocol( host )
1944        self.data_msg   = xdrlib.Packer()
1945        self.meta_msg   = xdrlib.Packer()
1946        self.socket     = socket.socket( socket.AF_INET, socket.SOCK_DGRAM )
1947
1948        if self.prot not in self.protocol:
1949
1950            raise ValueError( "Protocol must be one of: " + str( self.protocol ) )
1951
1952        if self.prot == 'multicast':
1953
1954            # Set multicast options
1955            #
1956            self.socket.setsockopt( socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 20 )
1957
1958        self.hostport   = ( host, int( port ) )
1959        self.slopestr   = 'both'
1960        self.tmax       = 60
1961
1962    def checkHostProtocol( self, ip ):
1963
1964        """Detect if a ip adress is a multicast address"""
1965
1966        MULTICAST_ADDRESS_MIN   = ( "224", "0", "0", "0" )
1967        MULTICAST_ADDRESS_MAX   = ( "239", "255", "255", "255" )
1968
1969        ip_fields               = ip.split( '.' )
1970
1971        if ip_fields >= MULTICAST_ADDRESS_MIN and ip_fields <= MULTICAST_ADDRESS_MAX:
1972
1973            return 'multicast'
1974        else:
1975            return 'udp'
1976
1977    def send( self, name, value, dmax, typestr = '', units = '' ):
1978
1979        if len( units ) == 0:
1980            units       = GMETRIC_DEFAULT_UNITS
1981
1982        if len( typestr ) == 0:
1983            typestr     = GMETRIC_DEFAULT_TYPE
1984
1985        (meta_msg, data_msg) = self.makexdr( name, value, typestr, units, self.slopestr, self.tmax, dmax )
1986
1987        meta_rt = self.socket.sendto( meta_msg, self.hostport )
1988        data_rt = self.socket.sendto( data_msg, self.hostport )
1989
1990        return ( meta_rt, data_rt )
1991
1992    def makexdr( self, name, value, typestr, unitstr, slopestr, tmax, dmax, group=None, spoof=None ):
1993
1994        hostname = "unset"
1995
1996        if slopestr not in self.slope:
1997
1998            raise ValueError( "Slope must be one of: " + str( self.slope.keys() ) )
1999
2000        if typestr not in self.type:
2001
2002            raise ValueError( "Type must be one of: " + str( self.type ) )
2003
2004        if len( name ) == 0:
2005
2006            raise ValueError( "Name must be non-empty" )
2007
2008        self.meta_msg.reset()
2009        self.meta_msg.pack_int( 128 )
2010
2011        if not spoof:
2012            self.meta_msg.pack_string( hostname )
2013        else:
2014            self.meta_msg.pack_string( spoof )
2015
2016        self.meta_msg.pack_string( name )
2017
2018        if not spoof:
2019            self.meta_msg.pack_int( 0 )
2020        else:
2021            self.meta_msg.pack_int( 1 )
2022           
2023        self.meta_msg.pack_string( typestr )
2024        self.meta_msg.pack_string( name )
2025        self.meta_msg.pack_string( unitstr )
2026        self.meta_msg.pack_int( self.slope[ slopestr ] )
2027        self.meta_msg.pack_uint( int( tmax ) )
2028        self.meta_msg.pack_uint( int( dmax ) )
2029
2030        if not group:
2031            self.meta_msg.pack_int( 0 )
2032        else:
2033            self.meta_msg.pack_int( 1 )
2034            self.meta_msg.pack_string( "GROUP" )
2035            self.meta_msg.pack_string( group )
2036
2037        self.data_msg.reset()
2038        self.data_msg.pack_int( 128+5 )
2039
2040        if not spoof:
2041            self.data_msg.pack_string( hostname )
2042        else:
2043            self.data_msg.pack_string( spoof )
2044
2045        self.data_msg.pack_string( name )
2046
2047        if not spoof:
2048            self.data_msg.pack_int( 0 )
2049        else:
2050            self.data_msg.pack_int( 1 )
2051
2052        self.data_msg.pack_string( "%s" )
2053        self.data_msg.pack_string( str( value ) )
2054
2055        return ( self.meta_msg.get_buffer(), self.data_msg.get_buffer() )
2056
2057def printTime( ):
2058
2059    """Print current time/date in human readable format for log/debug"""
2060
2061    return time.strftime("%a, %d %b %Y %H:%M:%S")
2062
2063def debug_msg( level, msg ):
2064
2065    """Print msg if at or above current debug level"""
2066
2067    global DAEMONIZE, DEBUG_LEVEL, SYSLOG_LEVEL
2068
2069    if (not DAEMONIZE and DEBUG_LEVEL >= level):
2070        sys.stderr.write( msg + '\n' )
2071
2072    if (DAEMONIZE and USE_SYSLOG and SYSLOG_LEVEL >= level):
2073        syslog.syslog( msg )
2074
2075def write_pidfile():
2076
2077    # Write pidfile if PIDFILE is set
2078    #
2079    if PIDFILE:
2080
2081        pid    = os.getpid()
2082
2083        pidfile    = open( PIDFILE, 'w' )
2084
2085        pidfile.write( str( pid ) )
2086        pidfile.close()
2087
2088def main():
2089
2090    """Application start"""
2091
2092    global PBSQuery, PBSError, lsfObject, pyslurm
2093    global SYSLOG_FACILITY, USE_SYSLOG, BATCH_API, DAEMONIZE, BATCH_SERVER
2094
2095    if not processArgs( sys.argv[1:] ):
2096
2097        sys.exit( 1 )
2098
2099    # Load appropriate DataGatherer depending on which BATCH_API is set
2100    # and any required modules for the Gatherer
2101    #
2102    if BATCH_API == 'pbs':
2103
2104        try:
2105            from PBSQuery import PBSQuery, PBSError
2106
2107        except ImportError, details:
2108
2109            print "FATAL ERROR: BATCH_API set to 'pbs' but python module 'pbs_python' is not found or installed"
2110            print details
2111            sys.exit( 1 )
2112
2113        gather = PbsDataGatherer()
2114
2115    elif BATCH_API == 'sge':
2116
2117        if BATCH_SERVER != 'localhost':
2118
2119            # Print and log, but continue execution
2120            err_msg = "WARNING: BATCH_API 'sge' ignores BATCH_SERVER (can only be 'localhost')"
2121            print err_msg
2122            debug_msg( 0, err_msg )
2123
2124        # Tested with SGE 6.0u11.
2125        #
2126        gather = SgeDataGatherer()
2127
2128    elif BATCH_API == 'lsf':
2129
2130        if BATCH_SERVER != 'localhost':
2131
2132            # Print and log, but continue execution
2133            err_msg = "WARNING: BATCH_API 'lsf' ignores BATCH_SERVER (can only be 'localhost')"
2134            print err_msg
2135            debug_msg( 0, err_msg )
2136
2137        try:
2138            from lsfObject import lsfObject
2139        except:
2140            print "FATAL ERROR: BATCH_API set to 'lsf' but python module 'lsfObject' is not found or installed"
2141            sys.exit( 1 )
2142
2143        gather = LsfDataGatherer()
2144
2145    elif BATCH_API == 'slurm':
2146
2147        if BATCH_SERVER != 'localhost':
2148
2149            # Print and log, but continue execution
2150            err_msg = "WARNING: BATCH_API 'slurm' ignores BATCH_SERVER (can only be 'localhost')"
2151            print err_msg
2152            debug_msg( 0, err_msg )
2153
2154        try:
2155            import pyslurm
2156        except:
2157            print "FATAL ERROR: BATCH_API set to 'slurm' but python module is not found or installed"
2158            sys.exit( 1 )
2159
2160        gather = SLURMDataGatherer()
2161
2162    else:
2163        print "FATAL ERROR: unknown BATCH_API '" + BATCH_API + "' is not supported"
2164
2165        sys.exit( 1 )
2166
2167    if( DAEMONIZE and USE_SYSLOG ):
2168
2169        syslog.openlog( 'jobmond', syslog.LOG_NOWAIT, SYSLOG_FACILITY )
2170
2171    if DAEMONIZE:
2172
2173        gather.daemon()
2174    else:
2175        gather.run()
2176
2177# wh00t? someone started me! :)
2178#
2179if __name__ == '__main__':
2180    main()
Note: See TracBrowser for help on using the repository browser.