Context Navigation

source: trunk/daemon/togad.py @ 148

Last change on this file since 148 was 146, checked in by bastiaans, 19 years ago

daemon/togad.py:

Fixed: rrdtool pipes were never closed, cuasing memory hogs and zombie processes

File size: 31.9 KB

Rev	Line
[3]	1	#!/usr/bin/env python
	2
[70]	3	import xml.sax
[71]	4	import xml.sax.handler
[5]	5	import socket
	6	import sys
[9]	7	import string
[12]	8	import os
	9	import os.path
[17]	10	import time
[36]	11	import threading
[38]	12	import random
[40]	13	from types import *
[60]	14	import DBClass
[3]	15
[8]	16	# Specify debugging level here;
	17	#
[38]	18	# 11 = XML: metrics
	19	# 10 = XML: host, cluster, grid, ganglia
	20	# 9 = RRD activity, gmetad config parsing
[40]	21	# 8 = RRD file activity
[84]	22	# 6 = SQL
[87]	23	# 1 = daemon threading
[8]	24	#
[102]	25	DEBUG_LEVEL = 1
[6]	26
[9]	27	# Where is the gmetad.conf located
	28	#
	29	GMETAD_CONF = '/etc/gmetad.conf'
	30
[60]	31	# Where to grab XML data from
	32	# Normally: local gmetad (port 8651)
	33	#
[63]	34	# Syntax: <hostname>:<port>
	35	#
	36	ARCHIVE_XMLSOURCE = "localhost:8651"
[60]	37
[9]	38	# List of data_source names to archive for
	39	#
[63]	40	# Syntax: [ "<clustername>", "<clustername>" ]
	41	#
[60]	42	ARCHIVE_DATASOURCES = [ "LISA Cluster" ]
[9]	43
[60]	44	# Where to store the archived rrd's
	45	#
	46	ARCHIVE_PATH = '/data/toga/rrds'
	47
[13]	48	# Amount of hours to store in one single archived .rrd
[9]	49	#
[43]	50	ARCHIVE_HOURS_PER_RRD = 12
[13]	51
[63]	52	# Toga's SQL dbase to use
[60]	53	#
[63]	54	# Syntax: <hostname>/<database>
[60]	55	#
[63]	56	TOGA_SQL_DBASE = "localhost/toga"
[60]	57
[22]	58	# Wether or not to run as a daemon in background
	59	#
	60	DAEMONIZE = 0
	61
[17]	62	######################
	63	# #
	64	# Configuration ends #
	65	# #
	66	######################
[13]	67
[60]	68	###
	69	# You'll only want to change anything below here unless you
[63]	70	# know what you are doing (i.e. your name is Ramon Bastiaans :D )
[60]	71	###
	72
[17]	73	# What XML data types not to store
[13]	74	#
[17]	75	UNSUPPORTED_ARCHIVE_TYPES = [ 'string' ]
[9]	76
[47]	77	# Maximum time (in seconds) a parsethread may run
	78	#
	79	PARSE_TIMEOUT = 60
	80
	81	# Maximum time (in seconds) a storethread may run
	82	#
	83	STORE_TIMEOUT = 360
	84
[8]	85	"""
	86	This is TOrque-GAnglia's data Daemon
	87	"""
	88
[84]	89	class DataSQLStore:
	90
	91	db_vars = None
	92	dbc = None
	93
	94	def __init__( self, hostname, database ):
	95
	96	self.db_vars = DBClass.InitVars(DataBaseName=database,
	97	User='root',
	98	Host=hostname,
	99	Password='',
	100	Dictionary='true')
	101
	102	try:
	103	self.dbc = DBClass.DB(self.db_vars)
	104	except DBClass.DBError, details:
	105	print 'Error in connection to db: %s' %details
	106	sys.exit(1)
	107
	108	def setDatabase(self, statement):
	109	ret = self.doDatabase('set', statement)
	110	return ret
	111
	112	def getDatabase(self, statement):
	113	ret = self.doDatabase('get', statement)
	114	return ret
	115
	116	def doDatabase(self, type, statement):
	117
	118	debug_msg( 6, 'doDatabase(): %s: %s' %(type, statement) )
	119	try:
	120	if type == 'set':
	121	result = self.dbc.Set( statement )
	122	self.dbc.Commit()
	123	elif type == 'get':
	124	result = self.dbc.Get( statement )
	125
	126	except DBClass.DBError, detail:
	127	operation = statement.split(' ')[0]
	128	print "%s operation on database failed while performing\n'%s'\n%s"\
	129	%(operation, statement, detail)
	130	sys.exit(1)
	131
	132	debug_msg( 6, 'doDatabase(): result: %s' %(result) )
	133	return result
	134
	135	def getNodeId( self, hostname ):
	136
[89]	137	id = self.getDatabase( "SELECT node_id FROM nodes WHERE node_hostname = '%s'" %hostname )
[84]	138
[89]	139	if len( id ) > 0:
[84]	140
[89]	141	id = id[0][0]
	142
[84]	143	return id
	144	else:
	145	return None
	146
	147	def getNodeIds( self, hostnames ):
	148
	149	ids = [ ]
	150
	151	for node in hostnames:
	152
	153	id = self.getNodeId( node )
	154
	155	if id:
	156	ids.append( id )
	157
	158	return ids
	159
	160	def getJobId( self, jobid ):
	161
	162	id = self.getDatabase( "SELECT job_id FROM jobs WHERE job_id = '%s'" %jobid )
	163
	164	if id:
[89]	165	id = id[0][0]
[84]	166
	167	return id
	168	else:
	169	return None
	170
	171	def addJob( self, job_id, jobattrs ):
	172
	173	if not self.getJobId( job_id ):
	174
	175	self.mutateJob( 'insert', job_id, jobattrs )
	176	else:
	177	self.mutateJob( 'update', job_id, jobattrs )
	178
	179	def mutateJob( self, action, job_id, jobattrs ):
	180
	181	job_values = [ 'name', 'queue', 'owner', 'requested_time', 'requested_memory', 'ppn', 'status', 'start_timestamp', 'stop_timestamp' ]
	182
	183	insert_col_str = 'job_id'
	184	insert_val_str = "'%s'" %job_id
	185	update_str = None
	186
	187	debug_msg( 6, 'mutateJob(): %s %s' %(action,job_id))
	188
[99]	189	ids = [ ]
	190
[84]	191	for valname, value in jobattrs.items():
	192
[96]	193	if valname in job_values and value != '':
[84]	194
	195	column_name = 'job_' + valname
	196
	197	if action == 'insert':
	198
	199	if not insert_col_str:
	200	insert_col_str = column_name
	201	else:
	202	insert_col_str = insert_col_str + ',' + column_name
	203
	204	if not insert_val_str:
	205	insert_val_str = value
	206	else:
	207	insert_val_str = insert_val_str + ",'%s'" %value
	208
	209	elif action == 'update':
	210
	211	if not update_str:
	212	update_str = "%s='%s'" %(column_name, value)
	213	else:
	214	update_str = update_str + ",%s='%s'" %(column_name, value)
	215
[90]	216	elif valname == 'nodes' and value:
[84]	217
[98]	218	ids = self.addNodes( value )
[86]	219	node_list = value
[84]	220
	221	if action == 'insert':
	222
	223	self.setDatabase( "INSERT INTO jobs ( %s ) VALUES ( %s )" %( insert_col_str, insert_val_str ) )
[86]	224
[99]	225	if len( ids ) > 0:
	226	self.addJobNodes( job_id, ids )
	227
[84]	228	elif action == 'update':
	229
[89]	230	self.setDatabase( "UPDATE jobs SET %s WHERE job_id=%s" %(update_str, job_id) )
[84]	231
	232	def addNodes( self, hostnames ):
	233
[98]	234	ids = [ ]
	235
[84]	236	for node in hostnames:
	237
	238	id = self.getNodeId( node )
	239
	240	if not id:
	241	self.setDatabase( "INSERT INTO nodes ( node_hostname ) VALUES ( '%s' )" %node )
[98]	242	id = self.getNodeId( node )
[84]	243
[98]	244	ids.append( id )
	245
	246	return ids
	247
[86]	248	def addJobNodes( self, jobid, nodes ):
	249
	250	for node in nodes:
	251	self.addJobNode( jobid, node )
	252
[84]	253	def addJobNode( self, jobid, nodeid ):
	254
	255	self.setDatabase( "INSERT INTO job_nodes (job_id,node_id) VALUES ( %s,%s )" %(jobid, nodeid) )
	256
	257	def storeJobInfo( self, jobid, jobattrs ):
	258
	259	self.addJob( jobid, jobattrs )
	260
[37]	261	class RRDMutator:
[63]	262	"""A class for performing RRD mutations"""
[37]	263
	264	binary = '/usr/bin/rrdtool'
	265
	266	def __init__( self, binary=None ):
[63]	267	"""Set alternate binary if supplied"""
[37]	268
	269	if binary:
	270	self.binary = binary
	271
	272	def create( self, filename, args ):
[63]	273	"""Create a new rrd with args"""
	274
[40]	275	return self.perform( 'create', '"' + filename + '"', args )
[37]	276
	277	def update( self, filename, args ):
[63]	278	"""Update a rrd with args"""
	279
[40]	280	return self.perform( 'update', '"' + filename + '"', args )
[37]	281
[42]	282	def grabLastUpdate( self, filename ):
[63]	283	"""Determine the last update time of filename rrd"""
[42]	284
	285	last_update = 0
	286
[53]	287	debug_msg( 8, self.binary + ' info "' + filename + '"' )
	288
[42]	289	for line in os.popen( self.binary + ' info "' + filename + '"' ).readlines():
	290
	291	if line.find( 'last_update') != -1:
	292
	293	last_update = line.split( ' = ' )[1]
	294
	295	if last_update:
	296	return last_update
	297	else:
	298	return 0
	299
[40]	300	def perform( self, action, filename, args ):
[63]	301	"""Perform action on rrd filename with args"""
[37]	302
	303	arg_string = None
	304
[40]	305	if type( args ) is not ListType:
	306	debug_msg( 8, 'Arguments needs to be of type List' )
	307	return 1
	308
[37]	309	for arg in args:
	310
	311	if not arg_string:
	312
	313	arg_string = arg
	314	else:
	315	arg_string = arg_string + ' ' + arg
	316
[54]	317	debug_msg( 8, self.binary + ' ' + action + ' ' + filename + ' ' + arg_string )
[37]	318
[146]	319	cmd = os.popen( self.binary + ' ' + action + ' ' + filename + ' ' + arg_string )
	320	lines = cmd.readlines()
	321	cmd.close()
[37]	322
[146]	323	for line in lines:
	324
[37]	325	if line.find( 'ERROR' ) != -1:
	326
	327	error_msg = string.join( line.split( ' ' )[1:] )
[40]	328	debug_msg( 8, error_msg )
[37]	329	return 1
	330
	331	return 0
	332
[78]	333	class XMLProcessor:
	334	"""Skeleton class for XML processor's"""
	335
	336	def daemon( self ):
	337	"""Run as daemon forever"""
	338
	339	# Fork the first child
	340	#
	341	pid = os.fork()
	342
	343	if pid > 0:
	344
	345	sys.exit(0) # end parent
	346
	347	# creates a session and sets the process group ID
	348	#
	349	os.setsid()
	350
	351	# Fork the second child
	352	#
	353	pid = os.fork()
	354
	355	if pid > 0:
	356
	357	sys.exit(0) # end parent
	358
	359	# Go to the root directory and set the umask
	360	#
	361	os.chdir('/')
	362	os.umask(0)
	363
	364	sys.stdin.close()
	365	sys.stdout.close()
[81]	366	sys.stderr.close()
[78]	367
	368	os.open('/dev/null', 0)
	369	os.dup(0)
	370	os.dup(0)
	371
	372	self.run()
	373
	374	def run( self ):
	375	"""Do main processing of XML here"""
	376
	377	pass
	378
	379	class TorqueXMLProcessor( XMLProcessor ):
	380	"""Main class for processing XML and acting with it"""
	381
	382	def __init__( self ):
	383	"""Setup initial XML connection and handlers"""
	384
	385	self.myXMLGatherer = XMLGatherer( ARCHIVE_XMLSOURCE.split( ':' )[0], ARCHIVE_XMLSOURCE.split( ':' )[1] )
	386	self.myXMLSource = self.myXMLGatherer.getFileObject()
	387	self.myXMLHandler = TorqueXMLHandler()
	388	self.myXMLError = XMLErrorHandler()
[87]	389	self.config = GangliaConfigParser( GMETAD_CONF )
[78]	390
	391	def run( self ):
	392	"""Main XML processing"""
	393
[102]	394	debug_msg( 1, printTime() + ' - torque_xml_thread(): started.' )
[87]	395
[78]	396	while( 1 ):
	397
	398	self.myXMLSource = self.myXMLGatherer.getFileObject()
[102]	399	debug_msg( 1, printTime() + ' - torque_xml_thread(): Parsing..' )
[78]	400	xml.sax.parse( self.myXMLSource, self.myXMLHandler, self.myXMLError )
[102]	401	debug_msg( 1, printTime() + ' - torque_xml_thread(): Done parsing.' )
	402	debug_msg( 1, printTime() + ' - torque_xml_thread(): Sleeping.. (%ss)' %(str( self.config.getLowestInterval() ) ) )
[87]	403	time.sleep( self.config.getLowestInterval() )
[78]	404
[71]	405	class TorqueXMLHandler( xml.sax.handler.ContentHandler ):
[63]	406	"""Parse Torque's jobinfo XML from our plugin"""
	407
[72]	408	jobAttrs = { }
	409
[84]	410	def __init__( self ):
	411
	412	self.ds = DataSQLStore( TOGA_SQL_DBASE.split( '/' )[0], TOGA_SQL_DBASE.split( '/' )[1] )
[98]	413	self.jobs_processed = [ ]
	414	self.jobs_to_store = [ ]
[84]	415
[63]	416	def startElement( self, name, attrs ):
	417	"""
	418	This XML will be all gmetric XML
	419	so there will be no specific start/end element
	420	just one XML statement with all info
	421	"""
	422
	423	heartbeat = 0
[73]	424
	425	jobinfo = { }
[63]	426
	427	if name == 'METRIC':
	428
[70]	429	metricname = attrs.get( 'NAME', "" )
[63]	430
	431	if metricname == 'TOGA-HEARTBEAT':
[70]	432	self.heartbeat = attrs.get( 'VAL', "" )
[63]	433
[70]	434	elif metricname.find( 'TOGA-JOB' ) != -1:
[63]	435
[72]	436	job_id = metricname.split( 'TOGA-JOB-' )[1]
[63]	437	val = attrs.get( 'VAL', "" )
	438
[96]	439	if not job_id in self.jobs_processed:
	440	self.jobs_processed.append( job_id )
	441
[73]	442	check_change = 0
	443
	444	if self.jobAttrs.has_key( job_id ):
	445	check_change = 1
	446
[63]	447	valinfo = val.split( ' ' )
	448
	449	for myval in valinfo:
	450
[84]	451	if len( myval.split( '=' ) ) > 1:
[63]	452
[84]	453	valname = myval.split( '=' )[0]
	454	value = myval.split( '=' )[1]
[70]	455
[84]	456	if valname == 'nodes':
	457	value = value.split( ';' )
[72]	458
[84]	459	jobinfo[ valname ] = value
	460
[73]	461	if check_change:
[100]	462	if self.jobinfoChanged( self.jobAttrs, job_id, jobinfo ) and self.jobAttrs[ job_id ]['status'] != 'E':
	463	self.jobAttrs[ job_id ]['stop_timestamp'] = ''
[82]	464	self.jobAttrs[ job_id ] = self.setJobAttrs( self.jobAttrs[ job_id ], jobinfo )
[84]	465	if not job_id in self.jobs_to_store:
	466	self.jobs_to_store.append( job_id )
	467
[102]	468	debug_msg( 6, 'jobinfo for job %s has changed' %job_id )
[73]	469	else:
	470	self.jobAttrs[ job_id ] = jobinfo
[84]	471
	472	if not job_id in self.jobs_to_store:
	473	self.jobs_to_store.append( job_id )
	474
[102]	475	debug_msg( 6, 'jobinfo for job %s has changed' %job_id )
[73]	476
[77]	477	def endDocument( self ):
[74]	478	"""When all metrics have gone, check if any jobs have finished"""
[72]	479
[74]	480	for jobid, jobinfo in self.jobAttrs.items():
	481
	482	# This is an old job, not in current jobinfo list anymore
	483	# it must have finished, since we _did_ get a new heartbeat
	484	#
[96]	485	mytime = int( jobinfo['reported'] ) + int( jobinfo['poll_interval'] )
[102]	486
[96]	487	if mytime < self.heartbeat and jobid not in self.jobs_processed and jobinfo['status'] == 'R':
[74]	488
[97]	489	if not jobid in self.jobs_processed:
[96]	490	self.jobs_processed.append( jobid )
	491
[74]	492	self.jobAttrs[ jobid ]['status'] = 'F'
[81]	493	self.jobAttrs[ jobid ]['stop_timestamp'] = str( int( jobinfo['reported'] ) + int( jobinfo['poll_interval'] ) )
[96]	494
[84]	495	if not jobid in self.jobs_to_store:
	496	self.jobs_to_store.append( jobid )
[74]	497
[102]	498	debug_msg( 1, printTime() + ' - torque_xml_thread(): Storing..' )
[87]	499
[84]	500	for jobid in self.jobs_to_store:
	501	self.ds.storeJobInfo( jobid, self.jobAttrs[ jobid ] )
	502
[102]	503	debug_msg( 1, printTime() + ' - torque_xml_thread(): Done storing.' )
[87]	504
[96]	505	self.jobs_processed = [ ]
[84]	506	self.jobs_to_store = [ ]
	507
[82]	508	def setJobAttrs( self, old, new ):
	509	"""
	510	Set new job attributes in old, but not lose existing fields
	511	if old attributes doesn't have those
	512	"""
	513
	514	for valname, value in new.items():
	515	old[ valname ] = value
	516
	517	return old
	518
	519
[73]	520	def jobinfoChanged( self, jobattrs, jobid, jobinfo ):
[74]	521	"""
	522	Check if jobinfo has changed from jobattrs[jobid]
	523	if it's report time is bigger than previous one
	524	and it is report time is recent (equal to heartbeat)
	525	"""
[72]	526
[87]	527	ignore_changes = [ 'reported' ]
	528
[73]	529	if jobattrs.has_key( jobid ):
	530
	531	for valname, value in jobinfo.items():
	532
[87]	533	if valname not in ignore_changes:
[73]	534
[87]	535	if jobattrs[ jobid ].has_key( valname ):
[73]	536
[87]	537	if value != jobattrs[ jobid ][ valname ]:
[73]	538
[87]	539	if jobinfo['reported'] > jobattrs[ jobid ][ 'reported' ] and jobinfo['reported'] == self.heartbeat:
	540	return 1
[73]	541
[87]	542	else:
	543	return 1
	544
[73]	545	return 0
	546
[71]	547	class GangliaXMLHandler( xml.sax.handler.ContentHandler ):
[63]	548	"""Parse Ganglia's XML"""
[3]	549
[33]	550	def __init__( self, config ):
[63]	551	"""Setup initial variables and gather info on existing rrd archive"""
	552
[33]	553	self.config = config
[35]	554	self.clusters = { }
[54]	555	debug_msg( 0, printTime() + ' - Checking existing toga rrd archive..' )
[44]	556	self.gatherClusters()
[54]	557	debug_msg( 0, printTime() + ' - Check done.' )
[33]	558
[44]	559	def gatherClusters( self ):
[63]	560	"""Find all existing clusters in archive dir"""
[44]	561
[45]	562	archive_dir = check_dir(ARCHIVE_PATH)
[44]	563
	564	hosts = [ ]
	565
	566	if os.path.exists( archive_dir ):
	567
	568	dirlist = os.listdir( archive_dir )
	569
	570	for item in dirlist:
	571
	572	clustername = item
	573
[60]	574	if not self.clusters.has_key( clustername ) and clustername in ARCHIVE_DATASOURCES:
[44]	575
	576	self.clusters[ clustername ] = RRDHandler( self.config, clustername )
	577
[6]	578	def startElement( self, name, attrs ):
[63]	579	"""Memorize appropriate data from xml start tags"""
[3]	580
[7]	581	if name == 'GANGLIA_XML':
[32]	582
	583	self.XMLSource = attrs.get( 'SOURCE', "" )
	584	self.gangliaVersion = attrs.get( 'VERSION', "" )
	585
[12]	586	debug_msg( 10, 'Found XML data: source %s version %s' %( self.XMLSource, self.gangliaVersion ) )
[6]	587
[7]	588	elif name == 'GRID':
[32]	589
	590	self.gridName = attrs.get( 'NAME', "" )
	591	self.time = attrs.get( 'LOCALTIME', "" )
	592
[12]	593	debug_msg( 10, '`-Grid found: %s' %( self.gridName ) )
[6]	594
[7]	595	elif name == 'CLUSTER':
[32]	596
	597	self.clusterName = attrs.get( 'NAME', "" )
	598	self.time = attrs.get( 'LOCALTIME', "" )
	599
[60]	600	if not self.clusters.has_key( self.clusterName ) and self.clusterName in ARCHIVE_DATASOURCES:
[32]	601
[34]	602	self.clusters[ self.clusterName ] = RRDHandler( self.config, self.clusterName )
[33]	603
[35]	604	debug_msg( 10, ' \|-Cluster found: %s' %( self.clusterName ) )
[6]	605
[60]	606	elif name == 'HOST' and self.clusterName in ARCHIVE_DATASOURCES:
[32]	607
	608	self.hostName = attrs.get( 'NAME', "" )
	609	self.hostIp = attrs.get( 'IP', "" )
	610	self.hostReported = attrs.get( 'REPORTED', "" )
	611
[12]	612	debug_msg( 10, ' \| \|-Host found: %s - ip %s reported %s' %( self.hostName, self.hostIp, self.hostReported ) )
[6]	613
[60]	614	elif name == 'METRIC' and self.clusterName in ARCHIVE_DATASOURCES:
[6]	615
[32]	616	type = attrs.get( 'TYPE', "" )
[6]	617
[32]	618	if type not in UNSUPPORTED_ARCHIVE_TYPES:
[3]	619
[32]	620	myMetric = { }
	621	myMetric['name'] = attrs.get( 'NAME', "" )
	622	myMetric['val'] = attrs.get( 'VAL', "" )
	623	myMetric['time'] = self.hostReported
[3]	624
[34]	625	self.clusters[ self.clusterName ].memMetric( self.hostName, myMetric )
[3]	626
[34]	627	debug_msg( 11, ' \| \| \|-metric: %s:%s' %( myMetric['name'], myMetric['val'] ) )
[6]	628
[34]	629	def storeMetrics( self ):
[63]	630	"""Store metrics of each cluster rrd handler"""
[9]	631
[34]	632	for clustername, rrdh in self.clusters.items():
[16]	633
[38]	634	ret = rrdh.storeMetrics()
[9]	635
[38]	636	if ret:
	637	debug_msg( 9, 'An error occured while storing metrics for cluster %s' %clustername )
	638	return 1
	639
	640	return 0
	641
[71]	642	class XMLErrorHandler( xml.sax.handler.ErrorHandler ):
	643
	644	def error( self, exception ):
	645	"""Recoverable error"""
	646
	647	debug_msg( 0, 'Recoverable error ' + str( exception ) )
	648
	649	def fatalError( self, exception ):
	650	"""Non-recoverable error"""
	651
	652	exception_str = str( exception )
	653
	654	# Ignore 'no element found' errors
	655	if exception_str.find( 'no element found' ) != -1:
	656	debug_msg( 1, 'No XML data found: probably socket not (re)connected.' )
	657	return 0
	658
	659	debug_msg( 0, 'Non-recoverable error ' + str( exception ) )
	660	sys.exit( 1 )
	661
	662	def warning( self, exception ):
	663	"""Warning"""
	664
	665	debug_msg( 0, 'Warning ' + str( exception ) )
	666
[78]	667	class XMLGatherer:
[63]	668	"""Setup a connection and file object to Ganglia's XML"""
[3]	669
[8]	670	s = None
[70]	671	fd = None
[8]	672
	673	def __init__( self, host, port ):
[63]	674	"""Store host and port for connection"""
[8]	675
[5]	676	self.host = host
	677	self.port = port
[33]	678	self.connect()
[70]	679	self.makeFileDescriptor()
[3]	680
[33]	681	def connect( self ):
[63]	682	"""Setup connection to XML source"""
[8]	683
	684	for res in socket.getaddrinfo( self.host, self.port, socket.AF_UNSPEC, socket.SOCK_STREAM ):
[32]	685
[5]	686	af, socktype, proto, canonname, sa = res
[32]	687
[5]	688	try:
[32]	689
[8]	690	self.s = socket.socket( af, socktype, proto )
[32]	691
[5]	692	except socket.error, msg:
[32]	693
[8]	694	self.s = None
[5]	695	continue
[32]	696
[5]	697	try:
[32]	698
[8]	699	self.s.connect( sa )
[32]	700
[5]	701	except socket.error, msg:
[32]	702
[70]	703	self.disconnect()
[5]	704	continue
[32]	705
[5]	706	break
[3]	707
[8]	708	if self.s is None:
[32]	709
[33]	710	debug_msg( 0, 'Could not open socket' )
	711	sys.exit( 1 )
[5]	712
[33]	713	def disconnect( self ):
[63]	714	"""Close socket"""
[33]	715
	716	if self.s:
[71]	717	self.s.shutdown( 2 )
[33]	718	self.s.close()
	719	self.s = None
	720
	721	def __del__( self ):
[63]	722	"""Kill the socket before we leave"""
[33]	723
	724	self.disconnect()
	725
[70]	726	def reconnect( self ):
	727	"""Reconnect"""
[33]	728
[38]	729	if self.s:
	730	self.disconnect()
[33]	731
[70]	732	self.connect()
[5]	733
[70]	734	def makeFileDescriptor( self ):
	735	"""Make file descriptor that points to our socket connection"""
	736
	737	self.reconnect()
	738
	739	if self.s:
	740	self.fd = self.s.makefile( 'r' )
	741
	742	def getFileObject( self ):
	743	"""Connect, and return a file object"""
	744
[78]	745	self.makeFileDescriptor()
	746
[70]	747	if self.fd:
	748	return self.fd
	749
[78]	750	class GangliaXMLProcessor( XMLProcessor ):
[63]	751	"""Main class for processing XML and acting with it"""
[5]	752
[33]	753	def __init__( self ):
[63]	754	"""Setup initial XML connection and handlers"""
[33]	755
	756	self.config = GangliaConfigParser( GMETAD_CONF )
	757
[78]	758	self.myXMLGatherer = XMLGatherer( ARCHIVE_XMLSOURCE.split( ':' )[0], ARCHIVE_XMLSOURCE.split( ':' )[1] )
[70]	759	self.myXMLSource = self.myXMLGatherer.getFileObject()
[78]	760	self.myXMLHandler = GangliaXMLHandler( self.config )
	761	self.myXMLError = XMLErrorHandler()
[73]	762
[9]	763	def run( self ):
[63]	764	"""Main XML processing; start a xml and storethread"""
[8]	765
[102]	766	xml_thread = threading.Thread( None, self.processXML, 'xmlthread' )
	767	store_thread = threading.Thread( None, self.storeMetrics, 'storethread' )
[22]	768
[36]	769	while( 1 ):
	770
[102]	771	if not xml_thread.isAlive():
[36]	772	# Gather XML at the same interval as gmetad
	773
	774	# threaded call to: self.processXML()
	775	#
[102]	776	xml_thread = threading.Thread( None, self.processXML, 'xml_thread' )
	777	xml_thread.start()
[36]	778
[102]	779	if not store_thread.isAlive():
[55]	780	# Store metrics every .. sec
[36]	781
[55]	782	# threaded call to: self.storeMetrics()
	783	#
[102]	784	store_thread = threading.Thread( None, self.storeMetrics, 'store_thread' )
	785	store_thread.start()
[36]	786
	787	# Just sleep a sec here, to prevent daemon from going mad. We're all threads here anyway
	788	time.sleep( 1 )
	789
[33]	790	def storeMetrics( self ):
[63]	791	"""Store metrics retained in memory to disk"""
[22]	792
[102]	793	debug_msg( 1, printTime() + ' - ganglia_store_thread(): started.' )
[39]	794
[63]	795	# Store metrics somewhere between every 360 and 640 seconds
[38]	796	#
[55]	797	STORE_INTERVAL = random.randint( 360, 640 )
[22]	798
[102]	799	store_metric_thread = threading.Thread( None, self.storeThread, 'store_metric_thread' )
	800	store_metric_thread.start()
[36]	801
[102]	802	debug_msg( 1, printTime() + ' - ganglia_store_thread(): Sleeping.. (%ss)' %STORE_INTERVAL )
[36]	803	time.sleep( STORE_INTERVAL )
[102]	804	debug_msg( 1, printTime() + ' - ganglia_store_thread(): Done sleeping.' )
[36]	805
[102]	806	if store_metric_thread.isAlive():
[36]	807
[102]	808	debug_msg( 1, printTime() + ' - ganglia_store_thread(): storemetricthread() still running, waiting to finish..' )
[136]	809	store_metric_thread.join( STORE_TIMEOUT ) # Maximum time is for storing thread to finish
[102]	810	debug_msg( 1, printTime() + ' - ganglia_store_thread(): Done waiting.' )
[36]	811
[102]	812	debug_msg( 1, printTime() + ' - ganglia_store_thread(): finished.' )
[36]	813
	814	return 0
	815
[39]	816	def storeThread( self ):
[63]	817	"""Actual metric storing thread"""
[39]	818
[102]	819	debug_msg( 1, printTime() + ' - ganglia_store_metric_thread(): started.' )
	820	debug_msg( 1, printTime() + ' - ganglia_store_metric_thread(): Storing data..' )
[78]	821	ret = self.myXMLHandler.storeMetrics()
[102]	822	debug_msg( 1, printTime() + ' - ganglia_store_metric_thread(): Done storing.' )
	823	debug_msg( 1, printTime() + ' - ganglia_store_metric_thread(): finished.' )
[39]	824
	825	return ret
	826
[8]	827	def processXML( self ):
[63]	828	"""Process XML"""
[8]	829
[87]	830	debug_msg( 1, printTime() + ' - xmlthread(): started.' )
[8]	831
[39]	832	parsethread = threading.Thread( None, self.parseThread, 'parsethread' )
[36]	833	parsethread.start()
	834
[102]	835	debug_msg( 1, printTime() + ' - ganglia_xml_thread(): Sleeping.. (%ss)' %self.config.getLowestInterval() )
[36]	836	time.sleep( float( self.config.getLowestInterval() ) )
[102]	837	debug_msg( 1, printTime() + ' - ganglia_xml_thread(): Done sleeping.' )
[36]	838
	839	if parsethread.isAlive():
	840
[102]	841	debug_msg( 1, printTime() + ' - ganglia_xml_thread(): parsethread() still running, waiting to finish..' )
[47]	842	parsethread.join( PARSE_TIMEOUT ) # Maximum time for XML thread to finish
[102]	843	debug_msg( 1, printTime() + ' - ganglia_xml_thread(): Done waiting.' )
[36]	844
[102]	845	debug_msg( 1, printTime() + ' - ganglia_xml_thread(): finished.' )
[36]	846
	847	return 0
	848
[39]	849	def parseThread( self ):
[63]	850	"""Actual parsing thread"""
[39]	851
[102]	852	debug_msg( 1, printTime() + ' - ganglia_parse_thread(): started.' )
	853	debug_msg( 1, printTime() + ' - ganglia_parse_thread(): Parsing XML..' )
[78]	854	self.myXMLSource = self.myXMLGatherer.getFileObject()
	855	ret = xml.sax.parse( self.myXMLSource, self.myXMLHandler, self.myXMLError )
[102]	856	debug_msg( 1, printTime() + ' - ganglia_parse_thread(): Done parsing.' )
	857	debug_msg( 1, printTime() + ' - ganglia_parse_thread(): finished.' )
[39]	858
	859	return ret
	860
[9]	861	class GangliaConfigParser:
	862
[34]	863	sources = [ ]
[9]	864
	865	def __init__( self, config ):
[63]	866	"""Parse some stuff from our gmetad's config, such as polling interval"""
[32]	867
[9]	868	self.config = config
	869	self.parseValues()
	870
[32]	871	def parseValues( self ):
[63]	872	"""Parse certain values from gmetad.conf"""
[9]	873
	874	readcfg = open( self.config, 'r' )
	875
	876	for line in readcfg.readlines():
	877
	878	if line.count( '"' ) > 1:
	879
[10]	880	if line.find( 'data_source' ) != -1 and line[0] != '#':
[9]	881
[11]	882	source = { }
	883	source['name'] = line.split( '"' )[1]
[9]	884	source_words = line.split( '"' )[2].split( ' ' )
	885
	886	for word in source_words:
	887
	888	valid_interval = 1
	889
	890	for letter in word:
[32]	891
[9]	892	if letter not in string.digits:
[32]	893
[9]	894	valid_interval = 0
	895
[10]	896	if valid_interval and len(word) > 0:
[32]	897
[9]	898	source['interval'] = word
[12]	899	debug_msg( 9, 'polling interval for %s = %s' %(source['name'], source['interval'] ) )
[33]	900
	901	# No interval found, use Ganglia's default
	902	if not source.has_key( 'interval' ):
	903	source['interval'] = 15
	904	debug_msg( 9, 'polling interval for %s defaulted to 15' %(source['name']) )
[32]	905
[33]	906	self.sources.append( source )
[9]	907
	908	def getInterval( self, source_name ):
[63]	909	"""Return interval for source_name"""
[32]	910
[9]	911	for source in self.sources:
[32]	912
[12]	913	if source['name'] == source_name:
[32]	914
[9]	915	return source['interval']
[32]	916
[9]	917	return None
	918
[34]	919	def getLowestInterval( self ):
[63]	920	"""Return the lowest interval of all clusters"""
[34]	921
	922	lowest_interval = 0
	923
	924	for source in self.sources:
	925
	926	if not lowest_interval or source['interval'] <= lowest_interval:
	927
	928	lowest_interval = source['interval']
	929
	930	# Return 15 when nothing is found, so that the daemon won't go insane with 0 sec delays
	931	if lowest_interval:
	932	return lowest_interval
	933	else:
	934	return 15
	935
[9]	936	class RRDHandler:
[63]	937	"""Class for handling RRD activity"""
[9]	938
[32]	939	myMetrics = { }
[40]	940	lastStored = { }
[47]	941	timeserials = { }
[36]	942	slot = None
[32]	943
[33]	944	def __init__( self, config, cluster ):
[63]	945	"""Setup initial variables"""
[78]	946
[44]	947	self.block = 0
[9]	948	self.cluster = cluster
[33]	949	self.config = config
[40]	950	self.slot = threading.Lock()
[37]	951	self.rrdm = RRDMutator()
[42]	952	self.gatherLastUpdates()
[9]	953
[42]	954	def gatherLastUpdates( self ):
[63]	955	"""Populate the lastStored list, containing timestamps of all last updates"""
[42]	956
	957	cluster_dir = '%s/%s' %( check_dir(ARCHIVE_PATH), self.cluster )
	958
	959	hosts = [ ]
	960
	961	if os.path.exists( cluster_dir ):
	962
[44]	963	dirlist = os.listdir( cluster_dir )
[42]	964
[44]	965	for dir in dirlist:
[42]	966
[44]	967	hosts.append( dir )
[42]	968
	969	for host in hosts:
	970
[47]	971	host_dir = cluster_dir + '/' + host
	972	dirlist = os.listdir( host_dir )
	973
	974	for dir in dirlist:
	975
	976	if not self.timeserials.has_key( host ):
	977
	978	self.timeserials[ host ] = [ ]
	979
	980	self.timeserials[ host ].append( dir )
	981
[42]	982	last_serial = self.getLastRrdTimeSerial( host )
	983	if last_serial:
	984
	985	metric_dir = cluster_dir + '/' + host + '/' + last_serial
	986	if os.path.exists( metric_dir ):
	987
[44]	988	dirlist = os.listdir( metric_dir )
[42]	989
[44]	990	for file in dirlist:
[42]	991
[44]	992	metricname = file.split( '.rrd' )[0]
[42]	993
[44]	994	if not self.lastStored.has_key( host ):
[42]	995
[44]	996	self.lastStored[ host ] = { }
[42]	997
[44]	998	self.lastStored[ host ][ metricname ] = self.rrdm.grabLastUpdate( metric_dir + '/' + file )
[42]	999
[32]	1000	def getClusterName( self ):
[63]	1001	"""Return clustername"""
	1002
[32]	1003	return self.cluster
	1004
	1005	def memMetric( self, host, metric ):
[63]	1006	"""Store metric from host in memory"""
[32]	1007
[34]	1008	if self.myMetrics.has_key( host ):
[32]	1009
[34]	1010	if self.myMetrics[ host ].has_key( metric['name'] ):
[32]	1011
[34]	1012	for mymetric in self.myMetrics[ host ][ metric['name'] ]:
[32]	1013
[34]	1014	if mymetric['time'] == metric['time']:
[32]	1015
[34]	1016	# Allready have this metric, abort
	1017	return 1
	1018	else:
	1019	self.myMetrics[ host ][ metric['name'] ] = [ ]
	1020	else:
[32]	1021	self.myMetrics[ host ] = { }
[34]	1022	self.myMetrics[ host ][ metric['name'] ] = [ ]
[32]	1023
[63]	1024	# Push new metric onto stack
	1025	# atomic code; only 1 thread at a time may access the stack
	1026
[53]	1027	# <ATOMIC>
[40]	1028	#
	1029	self.slot.acquire()
	1030
[32]	1031	self.myMetrics[ host ][ metric['name'] ].append( metric )
	1032
[40]	1033	self.slot.release()
[53]	1034	#
	1035	# </ATOMIC>
[40]	1036
[47]	1037	def makeUpdateList( self, host, metriclist ):
[63]	1038	"""
	1039	Make a list of update values for rrdupdate
	1040	but only those that we didn't store before
	1041	"""
[37]	1042
	1043	update_list = [ ]
[41]	1044	metric = None
[37]	1045
[47]	1046	while len( metriclist ) > 0:
[37]	1047
[53]	1048	metric = metriclist.pop( 0 )
[37]	1049
[53]	1050	if self.checkStoreMetric( host, metric ):
	1051	update_list.append( '%s:%s' %( metric['time'], metric['val'] ) )
[40]	1052
[37]	1053	return update_list
	1054
[49]	1055	def checkStoreMetric( self, host, metric ):
[63]	1056	"""Check if supplied metric if newer than last one stored"""
[40]	1057
	1058	if self.lastStored.has_key( host ):
	1059
[47]	1060	if self.lastStored[ host ].has_key( metric['name'] ):
[40]	1061
[47]	1062	if metric['time'] <= self.lastStored[ host ][ metric['name'] ]:
[40]	1063
[50]	1064	# This is old
[40]	1065	return 0
	1066
[50]	1067	return 1
	1068
[54]	1069	def memLastUpdate( self, host, metricname, metriclist ):
[63]	1070	"""
	1071	Memorize the time of the latest metric from metriclist
	1072	but only if it wasn't allready memorized
	1073	"""
[50]	1074
[54]	1075	if not self.lastStored.has_key( host ):
	1076	self.lastStored[ host ] = { }
	1077
[50]	1078	last_update_time = 0
	1079
	1080	for metric in metriclist:
	1081
[54]	1082	if metric['name'] == metricname:
[50]	1083
[54]	1084	if metric['time'] > last_update_time:
[50]	1085
[54]	1086	last_update_time = metric['time']
[40]	1087
[54]	1088	if self.lastStored[ host ].has_key( metricname ):
[52]	1089
[54]	1090	if last_update_time <= self.lastStored[ host ][ metricname ]:
[52]	1091	return 1
[40]	1092
[54]	1093	self.lastStored[ host ][ metricname ] = last_update_time
[52]	1094
[33]	1095	def storeMetrics( self ):
[63]	1096	"""
	1097	Store all metrics from memory to disk
	1098	and do it to the RRD's in appropriate timeperiod directory
	1099	"""
[33]	1100
	1101	for hostname, mymetrics in self.myMetrics.items():
	1102
	1103	for metricname, mymetric in mymetrics.items():
	1104
[53]	1105	metrics_to_store = [ ]
	1106
[63]	1107	# Pop metrics from stack for storing until none is left
	1108	# atomic code: only 1 thread at a time may access myMetrics
	1109
[53]	1110	# <ATOMIC>
[50]	1111	#
[47]	1112	self.slot.acquire()
[33]	1113
[54]	1114	while len( self.myMetrics[ hostname ][ metricname ] ) > 0:
[53]	1115
[54]	1116	if len( self.myMetrics[ hostname ][ metricname ] ) > 0:
	1117	metrics_to_store.append( self.myMetrics[ hostname ][ metricname ].pop( 0 ) )
[53]	1118
	1119	self.slot.release()
	1120	#
	1121	# </ATOMIC>
	1122
[47]	1123	# Create a mapping table, each metric to the period where it should be stored
	1124	#
[53]	1125	metric_serial_table = self.determineSerials( hostname, metricname, metrics_to_store )
[33]	1126
[50]	1127	update_rets = [ ]
	1128
[47]	1129	for period, pmetric in metric_serial_table.items():
	1130
[146]	1131	create_ret = self.createCheck( hostname, metricname, period )
[47]	1132
	1133	update_ret = self.update( hostname, metricname, period, pmetric )
	1134
	1135	if update_ret == 0:
	1136
	1137	debug_msg( 9, 'stored metric %s for %s' %( hostname, metricname ) )
	1138	else:
	1139	debug_msg( 9, 'metric update failed' )
	1140
[146]	1141	update_rets.append( create_ret )
[50]	1142	update_rets.append( update_ret )
[47]	1143
[50]	1144	if not (1) in update_rets:
	1145
[54]	1146	self.memLastUpdate( hostname, metricname, metrics_to_store )
[50]	1147
[17]	1148	def makeTimeSerial( self ):
[63]	1149	"""Generate a time serial. Seconds since epoch"""
[17]	1150
	1151	# Seconds since epoch
	1152	mytime = int( time.time() )
	1153
	1154	return mytime
	1155
[50]	1156	def makeRrdPath( self, host, metricname, timeserial ):
[63]	1157	"""Make a RRD location/path and filename"""
[17]	1158
[50]	1159	rrd_dir = '%s/%s/%s/%s' %( check_dir(ARCHIVE_PATH), self.cluster, host, timeserial )
	1160	rrd_file = '%s/%s.rrd' %( rrd_dir, metricname )
[17]	1161
	1162	return rrd_dir, rrd_file
	1163
[20]	1164	def getLastRrdTimeSerial( self, host ):
[63]	1165	"""Find the last timeserial (directory) for this host"""
[17]	1166
[19]	1167	newest_timeserial = 0
	1168
[47]	1169	for dir in self.timeserials[ host ]:
[32]	1170
[47]	1171	valid_dir = 1
[17]	1172
[47]	1173	for letter in dir:
	1174	if letter not in string.digits:
	1175	valid_dir = 0
[17]	1176
[47]	1177	if valid_dir:
	1178	timeserial = dir
	1179	if timeserial > newest_timeserial:
	1180	newest_timeserial = timeserial
[17]	1181
	1182	if newest_timeserial:
[18]	1183	return newest_timeserial
[17]	1184	else:
	1185	return 0
	1186
[47]	1187	def determinePeriod( self, host, check_serial ):
[63]	1188	"""Determine to which period (directory) this time(serial) belongs"""
[47]	1189
	1190	period_serial = 0
	1191
[56]	1192	if self.timeserials.has_key( host ):
[47]	1193
[56]	1194	for serial in self.timeserials[ host ]:
[47]	1195
[56]	1196	if check_serial >= serial and period_serial < serial:
[47]	1197
[56]	1198	period_serial = serial
	1199
[47]	1200	return period_serial
	1201
	1202	def determineSerials( self, host, metricname, metriclist ):
	1203	"""
	1204	Determine the correct serial and corresponding rrd to store
	1205	for a list of metrics
	1206	"""
	1207
	1208	metric_serial_table = { }
	1209
	1210	for metric in metriclist:
	1211
	1212	if metric['name'] == metricname:
	1213
	1214	period = self.determinePeriod( host, metric['time'] )
	1215
	1216	archive_secs = ARCHIVE_HOURS_PER_RRD * (60 * 60)
	1217
[49]	1218	if (int( metric['time'] ) - int( period ) ) > archive_secs:
[47]	1219
	1220	# This one should get it's own new period
	1221	period = metric['time']
[57]	1222
	1223	if not self.timeserials.has_key( host ):
	1224	self.timeserials[ host ] = [ ]
	1225
[50]	1226	self.timeserials[ host ].append( period )
[47]	1227
	1228	if not metric_serial_table.has_key( period ):
	1229
[49]	1230	metric_serial_table[ period ] = [ ]
[47]	1231
	1232	metric_serial_table[ period ].append( metric )
	1233
	1234	return metric_serial_table
	1235
[33]	1236	def createCheck( self, host, metricname, timeserial ):
[63]	1237	"""Check if an rrd allready exists for this metric, create if not"""
[9]	1238
[35]	1239	debug_msg( 9, 'rrdcreate: using timeserial %s for %s/%s' %( timeserial, host, metricname ) )
[47]	1240
[33]	1241	rrd_dir, rrd_file = self.makeRrdPath( host, metricname, timeserial )
[17]	1242
[9]	1243	if not os.path.exists( rrd_dir ):
[58]	1244
	1245	try:
	1246	os.makedirs( rrd_dir )
	1247
	1248	except OSError, msg:
	1249
	1250	if msg.find( 'File exists' ) != -1:
	1251
	1252	# Ignore exists errors
	1253	pass
	1254
	1255	else:
	1256
	1257	print msg
	1258	return
	1259
[14]	1260	debug_msg( 9, 'created dir %s' %( str(rrd_dir) ) )
[9]	1261
[14]	1262	if not os.path.exists( rrd_file ):
[9]	1263
[33]	1264	interval = self.config.getInterval( self.cluster )
[47]	1265	heartbeat = 8 * int( interval )
[9]	1266
[37]	1267	params = [ ]
[12]	1268
[37]	1269	params.append( '--step' )
	1270	params.append( str( interval ) )
[12]	1271
[37]	1272	params.append( '--start' )
[47]	1273	params.append( str( int( timeserial ) - 1 ) )
[12]	1274
[37]	1275	params.append( 'DS:sum:GAUGE:%d:U:U' %heartbeat )
	1276	params.append( 'RRA:AVERAGE:0.5:1:%s' %(ARCHIVE_HOURS_PER_RRD * 240) )
[13]	1277
[37]	1278	self.rrdm.create( str(rrd_file), params )
	1279
[14]	1280	debug_msg( 9, 'created rrd %s' %( str(rrd_file) ) )
	1281
[47]	1282	def update( self, host, metricname, timeserial, metriclist ):
[63]	1283	"""
	1284	Update rrd file for host with metricname
	1285	in directory timeserial with metriclist
	1286	"""
[9]	1287
[35]	1288	debug_msg( 9, 'rrdupdate: using timeserial %s for %s/%s' %( timeserial, host, metricname ) )
[9]	1289
[33]	1290	rrd_dir, rrd_file = self.makeRrdPath( host, metricname, timeserial )
[18]	1291
[47]	1292	update_list = self.makeUpdateList( host, metriclist )
[15]	1293
[41]	1294	if len( update_list ) > 0:
	1295	ret = self.rrdm.update( str(rrd_file), update_list )
[32]	1296
[41]	1297	if ret:
	1298	return 1
[27]	1299
[41]	1300	debug_msg( 9, 'updated rrd %s with %s' %( str(rrd_file), string.join( update_list ) ) )
[15]	1301
[36]	1302	return 0
	1303
[8]	1304	def main():
[63]	1305	"""Program startup"""
[8]	1306
[102]	1307	myTorqueProcessor = TorqueXMLProcessor()
	1308	myGangliaProcessor = GangliaXMLProcessor()
[8]	1309
[63]	1310	if DAEMONIZE:
[102]	1311	torque_xml_thread = threading.Thread( None, myTorqueProcessor.daemon, 'torque_proc_thread' )
	1312	ganglia_xml_thread = threading.Thread( None, myGangliaProcessor.daemon, 'ganglia_proc_thread' )
[63]	1313	else:
[102]	1314	torque_xml_thread = threading.Thread( None, myTorqueProcessor.run, 'torque_proc_thread' )
	1315	ganglia_xml_thread = threading.Thread( None, myGangliaProcessor.run, 'ganglia_proc_thread' )
[22]	1316
[102]	1317	torque_xml_thread.start()
	1318	ganglia_xml_thread.start()
[78]	1319
[81]	1320	# Global functions
	1321
[9]	1322	def check_dir( directory ):
[63]	1323	"""Check if directory is a proper directory. I.e.: Does _not_ end with a '/'"""
[9]	1324
	1325	if directory[-1] == '/':
	1326	directory = directory[:-1]
	1327
	1328	return directory
	1329
[12]	1330	def debug_msg( level, msg ):
[63]	1331	"""Only print msg if it is not below our debug level"""
[12]	1332
	1333	if (DEBUG_LEVEL >= level):
	1334	sys.stderr.write( msg + '\n' )
	1335
[46]	1336	def printTime( ):
[63]	1337	"""Print current time in human readable format"""
[46]	1338
	1339	return time.strftime("%a %d %b %Y %H:%M:%S")
	1340
[63]	1341	# Ooohh, someone started me! Let's go..
[9]	1342	if __name__ == '__main__':
	1343	main()

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

Download in other formats: