Context Navigation

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

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

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

Download in other formats: