Context Navigation

source: branches/0.4/jobmond/jobmond.py @ 692

Last change on this file since 692 was 692, checked in by ramonb, 11 years ago
going to require 3.4.0
Property svn:executable set to ``* Property svn:keywords set to `Id`
File size: 49.9 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 692 2013-03-20 10:15:35Z ramonb $
23	#
24
25	import sys, getopt, ConfigParser, time, os, socket, string, re
26	import xdrlib, socket, syslog, xml, xml.sax
27	from xml.sax.handler import feature_namespaces
28	from collections import deque
29
30	VERSION='0.4+SVN'
31
32	def usage( ver ):
33
34	print 'jobmond %s' %VERSION
35
36	if ver:
37	return 0
38
39	print
40	print 'Purpose:'
41	print ' The Job Monitoring Daemon (jobmond) reports batch jobs information and statistics'
42	print ' to Ganglia, which can be viewed with Job Monarch web frontend'
43	print
44	print 'Usage: jobmond [OPTIONS]'
45	print
46	print ' -c, --config=FILE The configuration file to use (default: /etc/jobmond.conf)'
47	print ' -p, --pidfile=FILE Use pid file to store the process id'
48	print ' -h, --help Print help and exit'
49	print ' -v, --version Print version and exit'
50	print
51
52	def processArgs( args ):
53
54	SHORT_L = 'p:hvc:'
55	LONG_L = [ 'help', 'config=', 'pidfile=', 'version' ]
56
57	global PIDFILE
58	PIDFILE = None
59
60	config_filename = '/etc/jobmond.conf'
61
62	try:
63
64	opts, args = getopt.getopt( args, SHORT_L, LONG_L )
65
66	except getopt.GetoptError, detail:
67
68	print detail
69	usage()
70	sys.exit( 1 )
71
72	for opt, value in opts:
73
74	if opt in [ '--config', '-c' ]:
75
76	config_filename = value
77
78	if opt in [ '--pidfile', '-p' ]:
79
80	PIDFILE = value
81
82	if opt in [ '--help', '-h' ]:
83
84	usage( False )
85	sys.exit( 0 )
86
87	if opt in [ '--version', '-v' ]:
88
89	usage( True )
90	sys.exit( 0 )
91
92	return loadConfig( config_filename )
93
94	# Fixme: This doesn't DTRT with commented-out bits of the file. E.g.
95	# it picked up a commented-out `mcast_join' and tried to use a
96	# multicast channel when it shouldn't have done.
97	class GangliaConfigParser:
98
99	def __init__( self, config_file ):
100
101	self.config_file = config_file
102
103	if not os.path.exists( self.config_file ):
104
105	debug_msg( 0, "FATAL ERROR: gmond config '" + self.config_file + "' not found!" )
106	sys.exit( 1 )
107
108	def removeQuotes( self, value ):
109
110	clean_value = value
111	clean_value = clean_value.replace( "'", "" )
112	clean_value = clean_value.replace( '"', '' )
113	clean_value = clean_value.strip()
114
115	return clean_value
116
117	def getVal( self, section, valname ):
118
119	cfg_fp = open( self.config_file )
120	section_start = False
121	section_found = False
122	value = None
123
124	for line in cfg_fp.readlines():
125
126	if line.find( section ) != -1:
127
128	section_found = True
129
130	if line.find( '{' ) != -1 and section_found:
131
132	section_start = True
133
134	if line.find( '}' ) != -1 and section_found:
135
136	section_start = False
137	section_found = False
138
139	if line.find( valname ) != -1 and section_start:
140
141	value = string.join( line.split( '=' )[1:], '' ).strip()
142
143	cfg_fp.close()
144
145	return value
146
147	def getInt( self, section, valname ):
148
149	value = self.getVal( section, valname )
150
151	if not value:
152	return False
153
154	value = self.removeQuotes( value )
155
156	return int( value )
157
158	def getStr( self, section, valname ):
159
160	value = self.getVal( section, valname )
161
162	if not value:
163	return False
164
165	value = self.removeQuotes( value )
166
167	return str( value )
168
169	def findGmetric():
170
171	for dir in os.path.expandvars( '$PATH' ).split( ':' ):
172
173	guess = '%s/%s' %( dir, 'gmetric' )
174
175	if os.path.exists( guess ):
176
177	return guess
178
179	return False
180
181	def loadConfig( filename ):
182
183	def getlist( cfg_string ):
184
185	my_list = [ ]
186
187	for item_txt in cfg_string.split( ',' ):
188
189	sep_char = None
190
191	item_txt = item_txt.strip()
192
193	for s_char in [ "'", '"' ]:
194
195	if item_txt.find( s_char ) != -1:
196
197	if item_txt.count( s_char ) != 2:
198
199	print 'Missing quote: %s' %item_txt
200	sys.exit( 1 )
201
202	else:
203
204	sep_char = s_char
205	break
206
207	if sep_char:
208
209	item_txt = item_txt.split( sep_char )[1]
210
211	my_list.append( item_txt )
212
213	return my_list
214
215	cfg = ConfigParser.ConfigParser()
216
217	cfg.read( filename )
218
219	global DEBUG_LEVEL, DAEMONIZE, BATCH_SERVER, BATCH_POLL_INTERVAL
220	global GMOND_CONF, DETECT_TIME_DIFFS, BATCH_HOST_TRANSLATE
221	global BATCH_API, QUEUE, GMETRIC_TARGET, USE_SYSLOG
222	global SYSLOG_LEVEL, SYSLOG_FACILITY, GMETRIC_BINARY
223
224	DEBUG_LEVEL = cfg.getint( 'DEFAULT', 'DEBUG_LEVEL' )
225
226	DAEMONIZE = cfg.getboolean( 'DEFAULT', 'DAEMONIZE' )
227
228	SYSLOG_LEVEL = -1
229	SYSLOG_FACILITY = None
230
231	try:
232	USE_SYSLOG = cfg.getboolean( 'DEFAULT', 'USE_SYSLOG' )
233
234	except ConfigParser.NoOptionError:
235
236	USE_SYSLOG = True
237
238	debug_msg( 0, 'ERROR: no option USE_SYSLOG found: assuming yes' )
239
240	if USE_SYSLOG:
241
242	try:
243	SYSLOG_LEVEL = cfg.getint( 'DEFAULT', 'SYSLOG_LEVEL' )
244
245	except ConfigParser.NoOptionError:
246
247	debug_msg( 0, 'ERROR: no option SYSLOG_LEVEL found: assuming level 0' )
248	SYSLOG_LEVEL = 0
249
250	try:
251
252	SYSLOG_FACILITY = eval( 'syslog.LOG_' + cfg.get( 'DEFAULT', 'SYSLOG_FACILITY' ) )
253
254	except ConfigParser.NoOptionError:
255
256	SYSLOG_FACILITY = syslog.LOG_DAEMON
257
258	debug_msg( 0, 'ERROR: no option SYSLOG_FACILITY found: assuming facility DAEMON' )
259
260	try:
261
262	BATCH_SERVER = cfg.get( 'DEFAULT', 'BATCH_SERVER' )
263
264	except ConfigParser.NoOptionError:
265
266	# Backwards compatibility for old configs
267	#
268
269	BATCH_SERVER = cfg.get( 'DEFAULT', 'TORQUE_SERVER' )
270	api_guess = 'pbs'
271
272	try:
273
274	BATCH_POLL_INTERVAL = cfg.getint( 'DEFAULT', 'BATCH_POLL_INTERVAL' )
275
276	except ConfigParser.NoOptionError:
277
278	# Backwards compatibility for old configs
279	#
280
281	BATCH_POLL_INTERVAL = cfg.getint( 'DEFAULT', 'TORQUE_POLL_INTERVAL' )
282	api_guess = 'pbs'
283
284	try:
285
286	GMOND_CONF = cfg.get( 'DEFAULT', 'GMOND_CONF' )
287
288	except ConfigParser.NoOptionError:
289
290	# Not specified: assume /etc/gmond.conf
291	#
292	GMOND_CONF = '/etc/gmond.conf'
293
294	ganglia_cfg = GangliaConfigParser( GMOND_CONF )
295
296	# Let's try to find the GMETRIC_TARGET ourselves first from GMOND_CONF
297	#
298	gmetric_dest_ip = ganglia_cfg.getStr( 'udp_send_channel', 'mcast_join' )
299
300	if not gmetric_dest_ip:
301
302	# Maybe unicast target then
303	#
304	gmetric_dest_ip = ganglia_cfg.getStr( 'udp_send_channel', 'host' )
305
306	gmetric_dest_port = ganglia_cfg.getStr( 'udp_send_channel', 'port' )
307
308	if gmetric_dest_ip and gmetric_dest_port:
309
310	GMETRIC_TARGET = '%s:%s' %( gmetric_dest_ip, gmetric_dest_port )
311	else:
312
313	debug_msg( 0, "WARNING: Can't parse udp_send_channel from: '%s'" %GMOND_CONF )
314
315	# Couldn't figure it out: let's see if it's in our jobmond.conf
316	#
317	try:
318
319	GMETRIC_TARGET = cfg.get( 'DEFAULT', 'GMETRIC_TARGET' )
320
321	# Guess not: now just give up
322	#
323	except ConfigParser.NoOptionError:
324
325	GMETRIC_TARGET = None
326
327	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!" )
328
329	gmetric_bin = findGmetric()
330
331	if gmetric_bin:
332
333	GMETRIC_BINARY = gmetric_bin
334	else:
335	debug_msg( 0, "WARNING: Can't find gmetric binary anywhere in $PATH" )
336
337	try:
338
339	GMETRIC_BINARY = cfg.get( 'DEFAULT', 'GMETRIC_BINARY' )
340
341	except ConfigParser.NoOptionError:
342
343	debug_msg( 0, "FATAL ERROR: GMETRIC_BINARY not set and not in $PATH" )
344	sys.exit( 1 )
345
346	DETECT_TIME_DIFFS = cfg.getboolean( 'DEFAULT', 'DETECT_TIME_DIFFS' )
347
348	BATCH_HOST_TRANSLATE = getlist( cfg.get( 'DEFAULT', 'BATCH_HOST_TRANSLATE' ) )
349
350	try:
351
352	BATCH_API = cfg.get( 'DEFAULT', 'BATCH_API' )
353
354	except ConfigParser.NoOptionError, detail:
355
356	if BATCH_SERVER and api_guess:
357
358	BATCH_API = api_guess
359	else:
360	debug_msg( 0, "FATAL ERROR: BATCH_API not set and can't make guess" )
361	sys.exit( 1 )
362
363	try:
364
365	QUEUE = getlist( cfg.get( 'DEFAULT', 'QUEUE' ) )
366
367	except ConfigParser.NoOptionError, detail:
368
369	QUEUE = None
370
371	return True
372
373	def fqdn_parts (fqdn):
374
375	"""Return pair of host and domain for fully-qualified domain name arg."""
376
377	parts = fqdn.split (".")
378
379	return (parts[0], string.join(parts[1:], "."))
380
381	METRIC_MAX_VAL_LEN = 900
382
383	class DataProcessor:
384
385	"""Class for processing of data"""
386
387	binary = None
388
389	def __init__( self, binary=None ):
390
391	"""Remember alternate binary location if supplied"""
392
393	global GMETRIC_BINARY, GMOND_CONF
394
395	if binary:
396	self.binary = binary
397
398	if not self.binary:
399	self.binary = GMETRIC_BINARY
400
401	# Timeout for XML
402	#
403	# From ganglia's documentation:
404	#
405	# 'A metric will be deleted DMAX seconds after it is received, and
406	# DMAX=0 means eternal life.'
407
408	self.dmax = str( int( int( BATCH_POLL_INTERVAL ) * 2 ) )
409
410	if GMOND_CONF:
411
412	incompatible = self.checkGmetricVersion()
413
414	if incompatible:
415
416	debug_msg( 0, 'Gmetric version not compatible, please upgrade to at least 3.4.0' )
417	sys.exit( 1 )
418
419	def checkGmetricVersion( self ):
420
421	"""
422	Check version of gmetric is at least 3.4.0
423	for the syntax we use
424	"""
425
426	global METRIC_MAX_VAL_LEN, GMETRIC_TARGET
427
428	incompatible = 0
429
430	gfp = os.popen( self.binary + ' --version' )
431	lines = gfp.readlines()
432
433	gfp.close()
434
435	for line in lines:
436
437	line = line.split( ' ' )
438
439	if len( line ) == 2 and str( line ).find( 'gmetric' ) != -1:
440
441	gmetric_version = line[1].split( '\n' )[0]
442
443	version_major = int( gmetric_version.split( '.' )[0] )
444	version_minor = int( gmetric_version.split( '.' )[1] )
445	version_patch = int( gmetric_version.split( '.' )[2] )
446
447	incompatible = 0
448
449	if version_major < 3:
450
451	incompatible = 1
452
453	elif version_major == 3:
454
455	if version_minor < 4:
456
457	incompatible = 1
458
459	else:
460
461	METRIC_MAX_VAL_LEN = 1400
462
463	return incompatible
464
465	def multicastGmetric( self, metricname, metricval, valtype='string', units='' ):
466
467	"""Call gmetric binary and multicast"""
468
469	cmd = self.binary
470
471	if GMETRIC_TARGET:
472
473	GMETRIC_TARGET_HOST = GMETRIC_TARGET.split( ':' )[0]
474	GMETRIC_TARGET_PORT = GMETRIC_TARGET.split( ':' )[1]
475
476	metric_debug = "[gmetric] name: %s - val: %s - dmax: %s" %( str( metricname ), str( metricval ), str( self.dmax ) )
477
478	debug_msg( 10, printTime() + ' ' + metric_debug)
479
480	gm = Gmetric( GMETRIC_TARGET_HOST, GMETRIC_TARGET_PORT )
481
482	gm.send( str( metricname ), str( metricval ), str( self.dmax ), valtype, units )
483
484	else:
485	try:
486	cmd = cmd + ' -c' + GMOND_CONF
487
488	except NameError:
489
490	debug_msg( 10, 'Assuming /etc/gmond.conf for gmetric cmd (omitting)' )
491
492	cmd = cmd + ' -n' + str( metricname )+ ' -v"' + str( metricval )+ '" -t' + str( valtype ) + ' -d' + str( self.dmax )
493
494	if len( units ) > 0:
495
496	cmd = cmd + ' -u"' + units + '"'
497
498	debug_msg( 10, printTime() + ' ' + cmd )
499
500	os.system( cmd )
501
502	class DataGatherer:
503
504	"""Skeleton class for batch system DataGatherer"""
505
506	def printJobs( self, jobs ):
507
508	"""Print a jobinfo overview"""
509
510	for name, attrs in self.jobs.items():
511
512	print 'job %s' %(name)
513
514	for name, val in attrs.items():
515
516	print '\t%s = %s' %( name, val )
517
518	def printJob( self, jobs, job_id ):
519
520	"""Print job with job_id from jobs"""
521
522	print 'job %s' %(job_id)
523
524	for name, val in jobs[ job_id ].items():
525
526	print '\t%s = %s' %( name, val )
527
528	def getAttr( self, attrs, name ):
529
530	"""Return certain attribute from dictionary, if exists"""
531
532	if attrs.has_key( name ):
533
534	return attrs[ name ]
535	else:
536	return ''
537
538	def jobDataChanged( self, jobs, job_id, attrs ):
539
540	"""Check if job with attrs and job_id in jobs has changed"""
541
542	if jobs.has_key( job_id ):
543
544	oldData = jobs[ job_id ]
545	else:
546	return 1
547
548	for name, val in attrs.items():
549
550	if oldData.has_key( name ):
551
552	if oldData[ name ] != attrs[ name ]:
553
554	return 1
555
556	else:
557	return 1
558
559	return 0
560
561	def submitJobData( self ):
562
563	"""Submit job info list"""
564
565	global BATCH_API
566
567	self.dp.multicastGmetric( 'MONARCH-HEARTBEAT', str( int( int( self.cur_time ) + int( self.timeoffset ) ) ) )
568
569	running_jobs = 0
570	queued_jobs = 0
571
572	# Count how many running/queued jobs we found
573	#
574	for jobid, jobattrs in self.jobs.items():
575
576	if jobattrs[ 'status' ] == 'Q':
577
578	queued_jobs += 1
579
580	elif jobattrs[ 'status' ] == 'R':
581
582	running_jobs += 1
583
584	# Report running/queued jobs as seperate metric for a nice RRD graph
585	#
586	self.dp.multicastGmetric( 'MONARCH-RJ', str( running_jobs ), 'uint32', 'jobs' )
587	self.dp.multicastGmetric( 'MONARCH-QJ', str( queued_jobs ), 'uint32', 'jobs' )
588
589	# Report down/offline nodes in batch (PBS only ATM)
590	#
591	if BATCH_API == 'pbs':
592
593	domain = fqdn_parts( socket.getfqdn() )[1]
594
595	downed_nodes = list()
596	offline_nodes = list()
597
598	l = ['state']
599
600	for name, node in self.pq.getnodes().items():
601
602	if ( node[ 'state' ].find( "down" ) != -1 ):
603
604	downed_nodes.append( name )
605
606	if ( node[ 'state' ].find( "offline" ) != -1 ):
607
608	offline_nodes.append( name )
609
610	downnodeslist = do_nodelist( downed_nodes )
611	offlinenodeslist = do_nodelist( offline_nodes )
612
613	down_str = 'nodes=%s domain=%s reported=%s' %( string.join( downnodeslist, ';' ), domain, str( int( int( self.cur_time ) + int( self.timeoffset ) ) ) )
614	offl_str = 'nodes=%s domain=%s reported=%s' %( string.join( offlinenodeslist, ';' ), domain, str( int( int( self.cur_time ) + int( self.timeoffset ) ) ) )
615	self.dp.multicastGmetric( 'MONARCH-DOWN' , down_str )
616	self.dp.multicastGmetric( 'MONARCH-OFFLINE', offl_str )
617
618	# Now let's spread the knowledge
619	#
620	for jobid, jobattrs in self.jobs.items():
621
622	# Make gmetric values for each job: respect max gmetric value length
623	#
624	gmetric_val = self.compileGmetricVal( jobid, jobattrs )
625	metric_increment = 0
626
627	# If we have more job info than max gmetric value length allows, split it up
628	# amongst multiple metrics
629	#
630	for val in gmetric_val:
631
632	self.dp.multicastGmetric( 'MONARCH-JOB-' + jobid + '-' + str(metric_increment), val )
633
634	# Increase follow number if this jobinfo is split up amongst more than 1 gmetric
635	#
636	metric_increment = metric_increment + 1
637
638	def compileGmetricVal( self, jobid, jobattrs ):
639
640	"""Create a val string for gmetric of jobinfo"""
641
642	gval_lists = [ ]
643	val_list = { }
644
645	for val_name, val_value in jobattrs.items():
646
647	# These are our own metric names, i.e.: status, start_timestamp, etc
648	#
649	val_list_names_len = len( string.join( val_list.keys() ) ) + len(val_list.keys())
650
651	# These are their corresponding values
652	#
653	val_list_vals_len = len( string.join( val_list.values() ) ) + len(val_list.values())
654
655	if val_name == 'nodes' and jobattrs['status'] == 'R':
656
657	node_str = None
658
659	for node in val_value:
660
661	if node_str:
662
663	node_str = node_str + ';' + node
664	else:
665	node_str = node
666
667	# Make sure if we add this new info, that the total metric's value length does not exceed METRIC_MAX_VAL_LEN
668	#
669	if (val_list_names_len + len(val_name) ) + (val_list_vals_len + len(node_str) ) > METRIC_MAX_VAL_LEN:
670
671	# It's too big, we need to make a new gmetric for the additional info
672	#
673	val_list[ val_name ] = node_str
674
675	gval_lists.append( val_list )
676
677	val_list = { }
678	node_str = None
679
680	val_list[ val_name ] = node_str
681
682	gval_lists.append( val_list )
683
684	val_list = { }
685
686	elif val_value != '':
687
688	# Make sure if we add this new info, that the total metric's value length does not exceed METRIC_MAX_VAL_LEN
689	#
690	if (val_list_names_len + len(val_name) ) + (val_list_vals_len + len(str(val_value)) ) > METRIC_MAX_VAL_LEN:
691
692	# It's too big, we need to make a new gmetric for the additional info
693	#
694	gval_lists.append( val_list )
695
696	val_list = { }
697
698	val_list[ val_name ] = val_value
699
700	if len( val_list ) > 0:
701
702	gval_lists.append( val_list )
703
704	str_list = [ ]
705
706	# Now append the value names and values together, i.e.: stop_timestamp=value, etc
707	#
708	for val_list in gval_lists:
709
710	my_val_str = None
711
712	for val_name, val_value in val_list.items():
713
714	if type(val_value) == list:
715
716	val_value = val_value.join( ',' )
717
718	if my_val_str:
719
720	try:
721	# fixme: It's getting
722	# ('nodes', None) items
723	my_val_str = my_val_str + ' ' + val_name + '=' + val_value
724	except:
725	pass
726
727	else:
728	my_val_str = val_name + '=' + val_value
729
730	str_list.append( my_val_str )
731
732	return str_list
733
734	def daemon( self ):
735
736	"""Run as daemon forever"""
737
738	# Fork the first child
739	#
740	pid = os.fork()
741	if pid > 0:
742	sys.exit(0) # end parent
743
744	# creates a session and sets the process group ID
745	#
746	os.setsid()
747
748	# Fork the second child
749	#
750	pid = os.fork()
751	if pid > 0:
752	sys.exit(0) # end parent
753
754	write_pidfile()
755
756	# Go to the root directory and set the umask
757	#
758	os.chdir('/')
759	os.umask(0)
760
761	sys.stdin.close()
762	sys.stdout.close()
763	sys.stderr.close()
764
765	os.open('/dev/null', os.O_RDWR)
766	os.dup2(0, 1)
767	os.dup2(0, 2)
768
769	self.run()
770
771	def run( self ):
772
773	"""Main thread"""
774
775	while ( 1 ):
776
777	self.getJobData()
778	self.submitJobData()
779	time.sleep( BATCH_POLL_INTERVAL )
780
781	# SGE code by Dave Love <fx@gnu.org>. Tested with SGE 6.0u8 and 6.0u11. May
782	# work with SGE 6.1 (else should be easily fixable), but definitely doesn't
783	# with 6.2. See also the fixmes.
784
785	class NoJobs (Exception):
786	"""Exception raised by empty job list in qstat output."""
787	pass
788
789	class SgeQstatXMLParser(xml.sax.handler.ContentHandler):
790	"""SAX handler for XML output from Sun Grid Engine's `qstat'."""
791
792	def __init__(self):
793	self.value = ""
794	self.joblist = []
795	self.job = {}
796	self.queue = ""
797	self.in_joblist = False
798	self.lrequest = False
799	self.eltq = deque()
800	xml.sax.handler.ContentHandler.__init__(self)
801
802	# The structure of the output is as follows (for SGE 6.0). It's
803	# similar for 6.1, but radically different for 6.2, and is
804	# undocumented generally. Unfortunately it's voluminous, and probably
805	# doesn't scale to large clusters/queues.
806
807	# <detailed_job_info xmlns:xsd="http://www.w3.org/2001/XMLSchema">
808	# <djob_info>
809	# <qmaster_response> <!-- job -->
810	# ...
811	# <JB_ja_template>
812	# <ulong_sublist>
813	# ... <!-- start_time, state ... -->
814	# </ulong_sublist>
815	# </JB_ja_template>
816	# <JB_ja_tasks>
817	# <ulong_sublist>
818	# ... <!-- task info
819	# </ulong_sublist>
820	# ...
821	# </JB_ja_tasks>
822	# ...
823	# </qmaster_response>
824	# </djob_info>
825	# <messages>
826	# ...
827
828	# NB. We might treat each task as a separate job, like
829	# straight qstat output, but the web interface expects jobs to
830	# be identified by integers, not, say, <job number>.<task>.
831
832	# So, I lied. If the job list is empty, we get invalid XML
833	# like this, which we need to defend against:
834
835	# <unknown_jobs xmlns:xsd="http://www.w3.org/2001/XMLSchema">
836	# <>
837	# <ST_name>*</ST_name>
838	# </>
839	# </unknown_jobs>
840
841	def startElement(self, name, attrs):
842	self.value = ""
843	if name == "djob_info": # job list
844	self.in_joblist = True
845	# The job container is "qmaster_response" in SGE 6.0
846	# and 6.1, but "element" in 6.2. This is only the very
847	# start of what's necessary for 6.2, though (sigh).
848	elif (name == "qmaster_response" or name == "element") \
849	and self.eltq[-1] == "djob_info": # job
850	self.job = {"job_state": "U", "slots": 0,
851	"nodes": [], "queued_timestamp": "",
852	"queued_timestamp": "", "queue": "",
853	"ppn": "0", "RN_max": 0,
854	# fixme in endElement
855	"requested_memory": 0, "requested_time": 0
856	}
857	self.joblist.append(self.job)
858	elif name == "qstat_l_requests": # resource request
859	self.lrequest = True
860	elif name == "unknown_jobs":
861	raise NoJobs
862	self.eltq.append (name)
863
864	def characters(self, ch):
865	self.value += ch
866
867	def endElement(self, name):
868	"""Snarf job elements contents into job dictionary.
869	Translate keys if appropriate."""
870
871	name_trans = {
872	"JB_job_number": "number",
873	"JB_job_name": "name", "JB_owner": "owner",
874	"queue_name": "queue", "JAT_start_time": "start_timestamp",
875	"JB_submission_time": "queued_timestamp"
876	}
877	value = self.value
878	self.eltq.pop ()
879
880	if name == "djob_info":
881	self.in_joblist = False
882	self.job = {}
883	elif name == "JAT_master_queue":
884	self.job["queue"] = value.split("@")[0]
885	elif name == "JG_qhostname":
886	if not (value in self.job["nodes"]):
887	self.job["nodes"].append(value)
888	elif name == "JG_slots": # slots in use
889	self.job["slots"] += int(value)
890	elif name == "RN_max": # requested slots (tasks or parallel)
891	self.job["RN_max"] = max (self.job["RN_max"],
892	int(value))
893	elif name == "JAT_state": # job state (bitwise or)
894	value = int (value)
895	# Status values from sge_jobL.h
896	#define JIDLE 0x00000000
897	#define JHELD 0x00000010
898	#define JMIGRATING 0x00000020
899	#define JQUEUED 0x00000040
900	#define JRUNNING 0x00000080
901	#define JSUSPENDED 0x00000100
902	#define JTRANSFERING 0x00000200
903	#define JDELETED 0x00000400
904	#define JWAITING 0x00000800
905	#define JEXITING 0x00001000
906	#define JWRITTEN 0x00002000
907	#define JSUSPENDED_ON_THRESHOLD 0x00010000
908	#define JFINISHED 0x00010000
909	if value & 0x80:
910	self.job["status"] = "R"
911	elif value & 0x40:
912	self.job["status"] = "Q"
913	else:
914	self.job["status"] = "O" # `other'
915	elif name == "CE_name" and self.lrequest and self.value in \
916	("h_cpu", "s_cpu", "cpu", "h_core", "s_core"):
917	# We're in a container for an interesting resource
918	# request; record which type.
919	self.lrequest = self.value
920	elif name == "CE_doubleval" and self.lrequest:
921	# if we're in a container for an interesting
922	# resource request, use the maxmimum of the hard
923	# and soft requests to record the requested CPU
924	# or core. Fixme: I'm not sure if this logic is
925	# right.
926	if self.lrequest in ("h_core", "s_core"):
927	self.job["requested_memory"] = \
928	max (float (value),
929	self.job["requested_memory"])
930	# Fixme: Check what cpu means, c.f [hs]_cpu.
931	elif self.lrequest in ("h_cpu", "s_cpu", "cpu"):
932	self.job["requested_time"] = \
933	max (float (value),
934	self.job["requested_time"])
935	elif name == "qstat_l_requests":
936	self.lrequest = False
937	elif self.job and self.in_joblist:
938	if name in name_trans:
939	name = name_trans[name]
940	self.job[name] = value
941
942	# Abstracted from PBS original.
943	# Fixme: Is it worth (or appropriate for PBS) sorting the result?
944	#
945	def do_nodelist( nodes ):
946
947	"""Translate node list as appropriate."""
948
949	nodeslist = [ ]
950	my_domain = fqdn_parts( socket.getfqdn() )[1]
951
952	for node in nodes:
953
954	host = node.split( '/' )[0] # not relevant for SGE
955	h, host_domain = fqdn_parts(host)
956
957	if host_domain == my_domain:
958
959	host = h
960
961	if nodeslist.count( host ) == 0:
962
963	for translate_pattern in BATCH_HOST_TRANSLATE:
964
965	if translate_pattern.find( '/' ) != -1:
966
967	translate_orig = \
968	translate_pattern.split( '/' )[1]
969	translate_new = \
970	translate_pattern.split( '/' )[2]
971	host = re.sub( translate_orig,
972	translate_new, host )
973	if not host in nodeslist:
974	nodeslist.append( host )
975	return nodeslist
976
977	class SgeDataGatherer(DataGatherer):
978
979	jobs = {}
980
981	def __init__( self ):
982	self.jobs = {}
983	self.timeoffset = 0
984	self.dp = DataProcessor()
985
986	def getJobData( self ):
987	"""Gather all data on current jobs in SGE"""
988
989	import popen2
990
991	self.cur_time = 0
992	queues = ""
993	if QUEUE: # only for specific queues
994	# Fixme: assumes queue names don't contain single
995	# quote or comma. Don't know what the SGE rules are.
996	queues = " -q '" + string.join (QUEUE, ",") + "'"
997	# Note the comment in SgeQstatXMLParser about scaling with
998	# this method of getting data. I haven't found better one.
999	# Output with args `-xml -ext -f -r' is easier to parse
1000	# in some ways, harder in others, but it doesn't provide
1001	# the submission time (at least SGE 6.0). The pipeline
1002	# into sed corrects bogus XML observed with a configuration
1003	# of SGE 6.0u8, which otherwise causes the parsing to hang.
1004	piping = popen2.Popen3("qstat -u '' -j '' -xml \| \
1005	sed -e 's/reported usage>/reported_usage>/g' -e 's;<\/JATASK:.>;;'" \
1006	+ queues, True)
1007	qstatparser = SgeQstatXMLParser()
1008	parse_err = 0
1009	try:
1010	xml.sax.parse(piping.fromchild, qstatparser)
1011	except NoJobs:
1012	pass
1013	except:
1014	parse_err = 1
1015	if piping.wait():
1016	debug_msg(10,
1017	"qstat error, skipping until next polling interval: "
1018	+ piping.childerr.readline())
1019	return None
1020	elif parse_err:
1021	debug_msg(10, "Bad XML output from qstat"())
1022	exit (1)
1023	for f in piping.fromchild, piping.tochild, piping.childerr:
1024	f.close()
1025	self.cur_time = time.time()
1026	jobs_processed = []
1027	for job in qstatparser.joblist:
1028	job_id = job["number"]
1029	if job["status"] in [ 'Q', 'R' ]:
1030	jobs_processed.append(job_id)
1031	if job["status"] == "R":
1032	job["nodes"] = do_nodelist (job["nodes"])
1033	# Fixme: why is job["nodes"] sometimes null?
1034	try:
1035	# Fixme: Is this sensible? The
1036	# PBS-type PPN isn't something you use
1037	# with SGE.
1038	job["ppn"] = float(job["slots"]) / \
1039	len(job["nodes"])
1040	except:
1041	job["ppn"] = 0
1042	if DETECT_TIME_DIFFS:
1043	# If a job start is later than our
1044	# current date, that must mean
1045	# the SGE server's time is later
1046	# than our local time.
1047	start_timestamp = \
1048	int (job["start_timestamp"])
1049	if start_timestamp > \
1050	int(self.cur_time) + \
1051	int(self.timeoffset):
1052
1053	self.timeoffset = \
1054	start_timestamp - \
1055	int(self.cur_time)
1056	else:
1057	# fixme: Note sure what this should be:
1058	job["ppn"] = job["RN_max"]
1059	job["nodes"] = "1"
1060
1061	myAttrs = {}
1062	for attr in ["name", "queue", "owner",
1063	"requested_time", "status",
1064	"requested_memory", "ppn",
1065	"start_timestamp", "queued_timestamp"]:
1066	myAttrs[attr] = str(job[attr])
1067	myAttrs["nodes"] = job["nodes"]
1068	myAttrs["reported"] = str(int(self.cur_time) + \
1069	int(self.timeoffset))
1070	myAttrs["domain"] = fqdn_parts(socket.getfqdn())[1]
1071	myAttrs["poll_interval"] = str(BATCH_POLL_INTERVAL)
1072
1073	if self.jobDataChanged(self.jobs, job_id, myAttrs) \
1074	and myAttrs["status"] in ["R", "Q"]:
1075	self.jobs[job_id] = myAttrs
1076	for id, attrs in self.jobs.items():
1077	if id not in jobs_processed:
1078	del self.jobs[id]
1079
1080	# LSF code by Mahmoud Hanafi <hanafim@users.sourceforge.nt>
1081	# Requres LSFObject http://sourceforge.net/projects/lsfobject
1082	#
1083	class LsfDataGatherer(DataGatherer):
1084
1085	"""This is the DataGatherer for LSf"""
1086
1087	global lsfObject
1088
1089	def __init__( self ):
1090
1091	self.jobs = { }
1092	self.timeoffset = 0
1093	self.dp = DataProcessor()
1094	self.initLsfQuery()
1095
1096	def _countDuplicatesInList( self, dupedList ):
1097
1098	countDupes = { }
1099
1100	for item in dupedList:
1101
1102	if not countDupes.has_key( item ):
1103
1104	countDupes[ item ] = 1
1105	else:
1106	countDupes[ item ] = countDupes[ item ] + 1
1107
1108	dupeCountList = [ ]
1109
1110	for item, count in countDupes.items():
1111
1112	dupeCountList.append( ( item, count ) )
1113
1114	return dupeCountList
1115	#
1116	#lst = ['I1','I2','I1','I3','I4','I4','I7','I7','I7','I7','I7']
1117	#print _countDuplicatesInList(lst)
1118	#[('I1', 2), ('I3', 1), ('I2', 1), ('I4', 2), ('I7', 5)]
1119	########################
1120
1121	def initLsfQuery( self ):
1122	self.pq = None
1123	self.pq = lsfObject.jobInfoEntObject()
1124
1125	def getJobData( self, known_jobs="" ):
1126	"""Gather all data on current jobs in LSF"""
1127	if len( known_jobs ) > 0:
1128	jobs = known_jobs
1129	else:
1130	jobs = { }
1131	joblist = {}
1132	joblist = self.pq.getJobInfo()
1133	nodelist = ''
1134
1135	self.cur_time = time.time()
1136
1137	jobs_processed = [ ]
1138
1139	for name, attrs in joblist.items():
1140	job_id = str(name)
1141	jobs_processed.append( job_id )
1142	name = self.getAttr( attrs, 'jobName' )
1143	queue = self.getAttr( self.getAttr( attrs, 'submit') , 'queue' )
1144	owner = self.getAttr( attrs, 'user' )
1145
1146	### THIS IS THE rLimit List index values
1147	#define LSF_RLIMIT_CPU 0 /* cpu time in milliseconds */
1148	#define LSF_RLIMIT_FSIZE 1 /* maximum file size */
1149	#define LSF_RLIMIT_DATA 2 /* data size */
1150	#define LSF_RLIMIT_STACK 3 /* stack size */
1151	#define LSF_RLIMIT_CORE 4 /* core file size */
1152	#define LSF_RLIMIT_RSS 5 /* resident set size */
1153	#define LSF_RLIMIT_NOFILE 6 /* open files */
1154	#define LSF_RLIMIT_OPEN_MAX 7 /* (from HP-UX) */
1155	#define LSF_RLIMIT_VMEM 8 /* maximum swap mem */
1156	#define LSF_RLIMIT_SWAP 8
1157	#define LSF_RLIMIT_RUN 9 /* max wall-clock time limit */
1158	#define LSF_RLIMIT_PROCESS 10 /* process number limit */
1159	#define LSF_RLIMIT_THREAD 11 /* thread number limit (introduced in LSF6.0) */
1160	#define LSF_RLIM_NLIMITS 12 /* number of resource limits */
1161
1162	requested_time = self.getAttr( self.getAttr( attrs, 'submit') , 'rLimits' )[9]
1163	if requested_time == -1:
1164	requested_time = ""
1165	requested_memory = self.getAttr( self.getAttr( attrs, 'submit') , 'rLimits' )[8]
1166	if requested_memory == -1:
1167	requested_memory = ""
1168	# This tries to get proc per node. We don't support this right now
1169	ppn = 0 #self.getAttr( self.getAttr( attrs, 'SubmitList') , 'numProessors' )
1170	requested_cpus = self.getAttr( self.getAttr( attrs, 'submit') , 'numProcessors' )
1171	if requested_cpus == None or requested_cpus == "":
1172	requested_cpus = 1
1173
1174	if QUEUE:
1175	for q in QUEUE:
1176	if q == queue:
1177	display_queue = 1
1178	break
1179	else:
1180	display_queue = 0
1181	continue
1182	if display_queue == 0:
1183	continue
1184
1185	runState = self.getAttr( attrs, 'status' )
1186	if runState == 4:
1187	status = 'R'
1188	else:
1189	status = 'Q'
1190	queued_timestamp = self.getAttr( attrs, 'submitTime' )
1191
1192	if status == 'R':
1193	start_timestamp = self.getAttr( attrs, 'startTime' )
1194	nodesCpu = dict(self._countDuplicatesInList(self.getAttr( attrs, 'exHosts' )))
1195	nodelist = nodesCpu.keys()
1196
1197	if DETECT_TIME_DIFFS:
1198
1199	# If a job start if later than our current date,
1200	# that must mean the Torque server's time is later
1201	# than our local time.
1202
1203	if int(start_timestamp) > int( int(self.cur_time) + int(self.timeoffset) ):
1204
1205	self.timeoffset = int( int(start_timestamp) - int(self.cur_time) )
1206
1207	elif status == 'Q':
1208	start_timestamp = ''
1209	count_mynodes = 0
1210	numeric_node = 1
1211	nodelist = ''
1212
1213	myAttrs = { }
1214	if name == "":
1215	myAttrs['name'] = "none"
1216	else:
1217	myAttrs['name'] = name
1218
1219	myAttrs[ 'owner' ] = owner
1220	myAttrs[ 'requested_time' ] = str(requested_time)
1221	myAttrs[ 'requested_memory' ] = str(requested_memory)
1222	myAttrs[ 'requested_cpus' ] = str(requested_cpus)
1223	myAttrs[ 'ppn' ] = str( ppn )
1224	myAttrs[ 'status' ] = status
1225	myAttrs[ 'start_timestamp' ] = str(start_timestamp)
1226	myAttrs[ 'queue' ] = str(queue)
1227	myAttrs[ 'queued_timestamp' ] = str(queued_timestamp)
1228	myAttrs[ 'reported' ] = str( int( int( self.cur_time ) + int( self.timeoffset ) ) )
1229	myAttrs[ 'nodes' ] = do_nodelist( nodelist )
1230	myAttrs[ 'domain' ] = fqdn_parts( socket.getfqdn() )[1]
1231	myAttrs[ 'poll_interval' ] = str(BATCH_POLL_INTERVAL)
1232
1233	if self.jobDataChanged( jobs, job_id, myAttrs ) and myAttrs['status'] in [ 'R', 'Q' ]:
1234	jobs[ job_id ] = myAttrs
1235
1236	debug_msg( 10, printTime() + ' job %s state changed' %(job_id) )
1237
1238	for id, attrs in jobs.items():
1239	if id not in jobs_processed:
1240	# This one isn't there anymore
1241	#
1242	del jobs[ id ]
1243	self.jobs=jobs
1244
1245
1246	class PbsDataGatherer( DataGatherer ):
1247
1248	"""This is the DataGatherer for PBS and Torque"""
1249
1250	global PBSQuery, PBSError
1251
1252	def __init__( self ):
1253
1254	"""Setup appropriate variables"""
1255
1256	self.jobs = { }
1257	self.timeoffset = 0
1258	self.dp = DataProcessor()
1259
1260	self.initPbsQuery()
1261
1262	def initPbsQuery( self ):
1263
1264	self.pq = None
1265
1266	if( BATCH_SERVER ):
1267
1268	self.pq = PBSQuery( BATCH_SERVER )
1269	else:
1270	self.pq = PBSQuery()
1271
1272	try:
1273	self.pq.old_data_structure()
1274
1275	except AttributeError:
1276
1277	# pbs_query is older
1278	#
1279	pass
1280
1281	def getJobData( self ):
1282
1283	"""Gather all data on current jobs in Torque"""
1284
1285	joblist = {}
1286	self.cur_time = 0
1287
1288	try:
1289	joblist = self.pq.getjobs()
1290	self.cur_time = time.time()
1291
1292	except PBSError, detail:
1293
1294	debug_msg( 10, "Caught PBS unavailable, skipping until next polling interval: " + str( detail ) )
1295	return None
1296
1297	jobs_processed = [ ]
1298
1299	for name, attrs in joblist.items():
1300	display_queue = 1
1301	job_id = name.split( '.' )[0]
1302
1303	name = self.getAttr( attrs, 'Job_Name' )
1304	queue = self.getAttr( attrs, 'queue' )
1305
1306	if QUEUE:
1307	for q in QUEUE:
1308	if q == queue:
1309	display_queue = 1
1310	break
1311	else:
1312	display_queue = 0
1313	continue
1314	if display_queue == 0:
1315	continue
1316
1317
1318	owner = self.getAttr( attrs, 'Job_Owner' ).split( '@' )[0]
1319	requested_time = self.getAttr( attrs, 'Resource_List.walltime' )
1320	requested_memory = self.getAttr( attrs, 'Resource_List.mem' )
1321
1322	mynoderequest = self.getAttr( attrs, 'Resource_List.nodes' )
1323
1324	ppn = ''
1325
1326	if mynoderequest.find( ':' ) != -1 and mynoderequest.find( 'ppn' ) != -1:
1327
1328	mynoderequest_fields = mynoderequest.split( ':' )
1329
1330	for mynoderequest_field in mynoderequest_fields:
1331
1332	if mynoderequest_field.find( 'ppn' ) != -1:
1333
1334	ppn = mynoderequest_field.split( 'ppn=' )[1]
1335
1336	status = self.getAttr( attrs, 'job_state' )
1337
1338	if status in [ 'Q', 'R' ]:
1339
1340	jobs_processed.append( job_id )
1341
1342	queued_timestamp = self.getAttr( attrs, 'ctime' )
1343
1344	if status == 'R':
1345
1346	start_timestamp = self.getAttr( attrs, 'mtime' )
1347	nodes = self.getAttr( attrs, 'exec_host' ).split( '+' )
1348
1349	nodeslist = do_nodelist( nodes )
1350
1351	if DETECT_TIME_DIFFS:
1352
1353	# If a job start if later than our current date,
1354	# that must mean the Torque server's time is later
1355	# than our local time.
1356
1357	if int( start_timestamp ) > int( int( self.cur_time ) + int( self.timeoffset ) ):
1358
1359	self.timeoffset = int( int(start_timestamp) - int(self.cur_time) )
1360
1361	elif status == 'Q':
1362
1363	# 'mynodequest' can be a string in the following syntax according to the
1364	# Torque Administator's manual:
1365	#
1366	# {<node_count> \| <hostname>}[:ppn=<ppn>][:<property>[:<property>]...][+ ...]
1367	# {<node_count> \| <hostname>}[:ppn=<ppn>][:<property>[:<property>]...][+ ...]
1368	# etc
1369	#
1370
1371	#
1372	# For now we only count the amount of nodes request and ignore properties
1373	#
1374
1375	start_timestamp = ''
1376	count_mynodes = 0
1377
1378	for node in mynoderequest.split( '+' ):
1379
1380	# Just grab the {node_count\|hostname} part and ignore properties
1381	#
1382	nodepart = node.split( ':' )[0]
1383
1384	# Let's assume a node_count value
1385	#
1386	numeric_node = 1
1387
1388	# Chop the value up into characters
1389	#
1390	for letter in nodepart:
1391
1392	# If this char is not a digit (0-9), this must be a hostname
1393	#
1394	if letter not in string.digits:
1395
1396	numeric_node = 0
1397
1398	# If this is a hostname, just count this as one (1) node
1399	#
1400	if not numeric_node:
1401
1402	count_mynodes = count_mynodes + 1
1403	else:
1404
1405	# If this a number, it must be the node_count
1406	# and increase our count with it's value
1407	#
1408	try:
1409	count_mynodes = count_mynodes + int( nodepart )
1410
1411	except ValueError, detail:
1412
1413	# When we arrive here I must be bugged or very confused
1414	# THIS SHOULD NOT HAPPEN!
1415	#
1416	debug_msg( 10, str( detail ) )
1417	debug_msg( 10, "Encountered weird node in Resources_List?!" )
1418	debug_msg( 10, 'nodepart = ' + str( nodepart ) )
1419	debug_msg( 10, 'job = ' + str( name ) )
1420	debug_msg( 10, 'attrs = ' + str( attrs ) )
1421
1422	nodeslist = str( count_mynodes )
1423	else:
1424	start_timestamp = ''
1425	nodeslist = ''
1426
1427	myAttrs = { }
1428
1429	myAttrs[ 'name' ] = str( name )
1430	myAttrs[ 'queue' ] = str( queue )
1431	myAttrs[ 'owner' ] = str( owner )
1432	myAttrs[ 'requested_time' ] = str( requested_time )
1433	myAttrs[ 'requested_memory' ] = str( requested_memory )
1434	myAttrs[ 'ppn' ] = str( ppn )
1435	myAttrs[ 'status' ] = str( status )
1436	myAttrs[ 'start_timestamp' ] = str( start_timestamp )
1437	myAttrs[ 'queued_timestamp' ] = str( queued_timestamp )
1438	myAttrs[ 'reported' ] = str( int( int( self.cur_time ) + int( self.timeoffset ) ) )
1439	myAttrs[ 'nodes' ] = nodeslist
1440	myAttrs[ 'domain' ] = fqdn_parts( socket.getfqdn() )[1]
1441	myAttrs[ 'poll_interval' ] = str( BATCH_POLL_INTERVAL )
1442
1443	if self.jobDataChanged( self.jobs, job_id, myAttrs ) and myAttrs['status'] in [ 'R', 'Q' ]:
1444
1445	self.jobs[ job_id ] = myAttrs
1446
1447	for id, attrs in self.jobs.items():
1448
1449	if id not in jobs_processed:
1450
1451	# This one isn't there anymore; toedeledoki!
1452	#
1453	del self.jobs[ id ]
1454
1455	#
1456	# Gmetric by Nick Galbreath - nickg(a.t)modp(d.o.t)com
1457	# Version 1.0 - 21-April2-2007
1458	# http://code.google.com/p/embeddedgmetric/
1459	#
1460	# Modified by: Ramon Bastiaans
1461	# For the Job Monarch Project, see: https://subtrac.sara.nl/oss/jobmonarch/
1462	#
1463	# added: DEFAULT_TYPE for Gmetric's
1464	# added: checkHostProtocol to determine if target is multicast or not
1465	# changed: allow default for Gmetric constructor
1466	# changed: allow defaults for all send() values except dmax
1467	#
1468
1469	GMETRIC_DEFAULT_TYPE = 'string'
1470	GMETRIC_DEFAULT_HOST = '127.0.0.1'
1471	GMETRIC_DEFAULT_PORT = '8649'
1472	GMETRIC_DEFAULT_UNITS = ''
1473
1474	class Gmetric:
1475
1476	global GMETRIC_DEFAULT_HOST, GMETRIC_DEFAULT_PORT
1477
1478	slope = { 'zero' : 0, 'positive' : 1, 'negative' : 2, 'both' : 3, 'unspecified' : 4 }
1479	type = ( '', 'string', 'uint16', 'int16', 'uint32', 'int32', 'float', 'double', 'timestamp' )
1480	protocol = ( 'udp', 'multicast' )
1481
1482	def __init__( self, host=GMETRIC_DEFAULT_HOST, port=GMETRIC_DEFAULT_PORT ):
1483
1484	global GMETRIC_DEFAULT_TYPE
1485
1486	self.prot = self.checkHostProtocol( host )
1487	self.msg = xdrlib.Packer()
1488	self.socket = socket.socket( socket.AF_INET, socket.SOCK_DGRAM )
1489
1490	if self.prot not in self.protocol:
1491
1492	raise ValueError( "Protocol must be one of: " + str( self.protocol ) )
1493
1494	if self.prot == 'multicast':
1495
1496	# Set multicast options
1497	#
1498	self.socket.setsockopt( socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 20 )
1499
1500	self.hostport = ( host, int( port ) )
1501	self.slopestr = 'both'
1502	self.tmax = 60
1503
1504	def checkHostProtocol( self, ip ):
1505
1506	"""Detect if a ip adress is a multicast address"""
1507
1508	MULTICAST_ADDRESS_MIN = ( "224", "0", "0", "0" )
1509	MULTICAST_ADDRESS_MAX = ( "239", "255", "255", "255" )
1510
1511	ip_fields = ip.split( '.' )
1512
1513	if ip_fields >= MULTICAST_ADDRESS_MIN and ip_fields <= MULTICAST_ADDRESS_MAX:
1514
1515	return 'multicast'
1516	else:
1517	return 'udp'
1518
1519	def send( self, name, value, dmax, typestr = '', units = '' ):
1520
1521	if len( units ) == 0:
1522	units = GMETRIC_DEFAULT_UNITS
1523
1524	if len( typestr ) == 0:
1525	typestr = GMETRIC_DEFAULT_TYPE
1526
1527	msg = self.makexdr( name, value, typestr, units, self.slopestr, self.tmax, dmax )
1528
1529	return self.socket.sendto( msg, self.hostport )
1530
1531	def makexdr( self, name, value, typestr, unitstr, slopestr, tmax, dmax ):
1532
1533	if slopestr not in self.slope:
1534
1535	raise ValueError( "Slope must be one of: " + str( self.slope.keys() ) )
1536
1537	if typestr not in self.type:
1538
1539	raise ValueError( "Type must be one of: " + str( self.type ) )
1540
1541	if len( name ) == 0:
1542
1543	raise ValueError( "Name must be non-empty" )
1544
1545	self.msg.reset()
1546	self.msg.pack_int( 0 )
1547	self.msg.pack_string( typestr )
1548	self.msg.pack_string( name )
1549	self.msg.pack_string( str( value ) )
1550	self.msg.pack_string( unitstr )
1551	self.msg.pack_int( self.slope[ slopestr ] )
1552	self.msg.pack_uint( int( tmax ) )
1553	self.msg.pack_uint( int( dmax ) )
1554
1555	return self.msg.get_buffer()
1556
1557	def printTime( ):
1558
1559	"""Print current time/date in human readable format for log/debug"""
1560
1561	return time.strftime("%a, %d %b %Y %H:%M:%S")
1562
1563	def debug_msg( level, msg ):
1564
1565	"""Print msg if at or above current debug level"""
1566
1567	global DAEMONIZE, DEBUG_LEVEL, SYSLOG_LEVEL
1568
1569	if (not DAEMONIZE and DEBUG_LEVEL >= level):
1570	sys.stderr.write( msg + '\n' )
1571
1572	if (DAEMONIZE and USE_SYSLOG and SYSLOG_LEVEL >= level):
1573	syslog.syslog( msg )
1574
1575	def write_pidfile():
1576
1577	# Write pidfile if PIDFILE is set
1578	#
1579	if PIDFILE:
1580
1581	pid = os.getpid()
1582
1583	pidfile = open( PIDFILE, 'w' )
1584
1585	pidfile.write( str( pid ) )
1586	pidfile.close()
1587
1588	def main():
1589
1590	"""Application start"""
1591
1592	global PBSQuery, PBSError, lsfObject
1593	global SYSLOG_FACILITY, USE_SYSLOG, BATCH_API, DAEMONIZE
1594
1595	if not processArgs( sys.argv[1:] ):
1596
1597	sys.exit( 1 )
1598
1599	# Load appropriate DataGatherer depending on which BATCH_API is set
1600	# and any required modules for the Gatherer
1601	#
1602	if BATCH_API == 'pbs':
1603
1604	try:
1605	from PBSQuery import PBSQuery, PBSError
1606
1607	except ImportError:
1608
1609	debug_msg( 0, "FATAL ERROR: BATCH_API set to 'pbs' but python module 'pbs_python' is not installed" )
1610	sys.exit( 1 )
1611
1612	gather = PbsDataGatherer()
1613
1614	elif BATCH_API == 'sge':
1615
1616	# Tested with SGE 6.0u11.
1617	#
1618	gather = SgeDataGatherer()
1619
1620	elif BATCH_API == 'lsf':
1621
1622	try:
1623	from lsfObject import lsfObject
1624	except:
1625	debug_msg(0, "fatal error: BATCH_API set to 'lsf' but python module is not found or installed")
1626	sys.exit( 1)
1627
1628	gather = LsfDataGatherer()
1629
1630	else:
1631	debug_msg( 0, "FATAL ERROR: unknown BATCH_API '" + BATCH_API + "' is not supported" )
1632
1633	sys.exit( 1 )
1634
1635	if( DAEMONIZE and USE_SYSLOG ):
1636
1637	syslog.openlog( 'jobmond', syslog.LOG_NOWAIT, SYSLOG_FACILITY )
1638
1639	if DAEMONIZE:
1640
1641	gather.daemon()
1642	else:
1643	gather.run()
1644
1645	# wh00t? someone started me! :)
1646	#
1647	if __name__ == '__main__':
1648	main()

Note: See TracBrowser for help on using the repository browser.

Download in other formats: