source: trunk/jobarchived/jobarchived.py @ 214

Last change on this file since 214 was 214, checked in by bastiaans, 16 years ago

jobarchived/jobarchived.conf:

  • import of config

jobarchived/jobarchived.py:

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