source: trunk/daemon/togad.py @ 40

Last change on this file since 40 was 40, checked in by bastiaans, 18 years ago

daemon/togad.py:

  • Changed thread locking scheme
  • Added function to remember last stored timestamp so that we will not try to update a rrd more than once with same timestamp
File size: 17.1 KB
Line 
1#!/usr/bin/env python
2
3from xml.sax import make_parser
4from xml.sax.handler import ContentHandler
5import socket
6import sys
7import string
8import os
9import os.path
10import time
11import re
12import threading
13import mutex
14import random
15from types import *
16
17# Specify debugging level here;
18#
19# 11 = XML: metrics
20# 10 = XML: host, cluster, grid, ganglia
21# 9  = RRD activity, gmetad config parsing
22# 8  = RRD file activity
23# 7  = daemon threading
24#
25DEBUG_LEVEL = 9
26
27# Where is the gmetad.conf located
28#
29GMETAD_CONF = '/etc/gmetad.conf'
30
31# Where to store the archived rrd's
32#
33ARCHIVE_PATH = '/data/toga/rrds'
34
35# List of data_source names to archive for
36#
37ARCHIVE_SOURCES = [ "LISA Cluster" ]
38
39# Amount of hours to store in one single archived .rrd
40#
41ARCHIVE_HOURS_PER_RRD = 1
42
43# Wether or not to run as a daemon in background
44#
45DAEMONIZE = 0
46
47######################
48#                    #
49# Configuration ends #
50#                    #
51######################
52
53# What XML data types not to store
54#
55UNSUPPORTED_ARCHIVE_TYPES = [ 'string' ]
56
57"""
58This is TOrque-GAnglia's data Daemon
59"""
60
61class RRDMutator:
62        "A class for handling .rrd mutations"
63
64        binary = '/usr/bin/rrdtool'
65
66        def __init__( self, binary=None ):
67
68                if binary:
69                        self.binary = binary
70
71        def create( self, filename, args ):
72                return self.perform( 'create', '"' + filename + '"', args )
73
74        def update( self, filename, args ):
75                return self.perform( 'update', '"' + filename + '"', args )
76
77        def perform( self, action, filename, args ):
78
79                arg_string = None
80
81                if type( args ) is not ListType:
82                        debug_msg( 8, 'Arguments needs to be of type List' )
83                        return 1
84
85                for arg in args:
86
87                        if not arg_string:
88
89                                arg_string = arg
90                        else:
91                                arg_string = arg_string + ' ' + arg
92
93                debug_msg( 8, 'rrdm.perform(): ' + self.binary + ' ' + action + ' ' + filename + ' ' + arg_string  )
94
95                for line in os.popen( self.binary + ' ' + action + ' ' + filename + ' ' + arg_string ).readlines():
96
97                        print line
98
99                        if line.find( 'ERROR' ) != -1:
100
101                                error_msg = string.join( line.split( ' ' )[1:] )
102                                debug_msg( 8, error_msg )
103                                return 1
104
105                return 0
106
107class GangliaXMLHandler( ContentHandler ):
108        "Parse Ganglia's XML"
109
110        def __init__( self, config ):
111                self.config = config
112                self.clusters = { }
113
114        def startElement( self, name, attrs ):
115                "Store appropriate data from xml start tags"
116
117                if name == 'GANGLIA_XML':
118
119                        self.XMLSource = attrs.get( 'SOURCE', "" )
120                        self.gangliaVersion = attrs.get( 'VERSION', "" )
121
122                        debug_msg( 10, 'Found XML data: source %s version %s' %( self.XMLSource, self.gangliaVersion ) )
123
124                elif name == 'GRID':
125
126                        self.gridName = attrs.get( 'NAME', "" )
127                        self.time = attrs.get( 'LOCALTIME', "" )
128
129                        debug_msg( 10, '`-Grid found: %s' %( self.gridName ) )
130
131                elif name == 'CLUSTER':
132
133                        self.clusterName = attrs.get( 'NAME', "" )
134                        self.time = attrs.get( 'LOCALTIME', "" )
135
136                        if not self.clusters.has_key( self.clusterName ) and self.clusterName in ARCHIVE_SOURCES:
137
138                                self.clusters[ self.clusterName ] = RRDHandler( self.config, self.clusterName )
139
140                                debug_msg( 10, ' |-Cluster found: %s' %( self.clusterName ) )
141
142                elif name == 'HOST' and self.clusterName in ARCHIVE_SOURCES:     
143
144                        self.hostName = attrs.get( 'NAME', "" )
145                        self.hostIp = attrs.get( 'IP', "" )
146                        self.hostReported = attrs.get( 'REPORTED', "" )
147
148                        debug_msg( 10, ' | |-Host found: %s - ip %s reported %s' %( self.hostName, self.hostIp, self.hostReported ) )
149
150                elif name == 'METRIC' and self.clusterName in ARCHIVE_SOURCES:
151
152                        type = attrs.get( 'TYPE', "" )
153
154                        if type not in UNSUPPORTED_ARCHIVE_TYPES:
155
156                                myMetric = { }
157                                myMetric['name'] = attrs.get( 'NAME', "" )
158                                myMetric['val'] = attrs.get( 'VAL', "" )
159                                myMetric['time'] = self.hostReported
160
161                                self.clusters[ self.clusterName ].memMetric( self.hostName, myMetric )
162
163                                debug_msg( 11, ' | | |-metric: %s:%s' %( myMetric['name'], myMetric['val'] ) )
164
165        def storeMetrics( self ):
166
167                for clustername, rrdh in self.clusters.items():
168
169                        ret = rrdh.storeMetrics()
170
171                        if ret:
172                                debug_msg( 9, 'An error occured while storing metrics for cluster %s' %clustername )
173                                return 1
174
175                return 0
176
177class GangliaXMLGatherer:
178        "Setup a connection and file object to Ganglia's XML"
179
180        s = None
181
182        def __init__( self, host, port ):
183                "Store host and port for connection"
184
185                self.host = host
186                self.port = port
187                self.connect()
188
189        def connect( self ):
190                "Setup connection to XML source"
191
192                for res in socket.getaddrinfo( self.host, self.port, socket.AF_UNSPEC, socket.SOCK_STREAM ):
193
194                        af, socktype, proto, canonname, sa = res
195
196                        try:
197
198                                self.s = socket.socket( af, socktype, proto )
199
200                        except socket.error, msg:
201
202                                self.s = None
203                                continue
204
205                        try:
206
207                                self.s.connect( sa )
208
209                        except socket.error, msg:
210
211                                self.s.close()
212                                self.s = None
213                                continue
214
215                        break
216
217                if self.s is None:
218
219                        debug_msg( 0, 'Could not open socket' )
220                        sys.exit( 1 )
221
222        def disconnect( self ):
223                "Close socket"
224
225                if self.s:
226                        self.s.close()
227                        self.s = None
228
229        def __del__( self ):
230                "Kill the socket before we leave"
231
232                self.disconnect()
233
234        def getFileObject( self ):
235                "Connect, and return a file object"
236
237                if self.s:
238                        # Apearantly, only data is received when a connection is made
239                        # therefor, disconnect and connect
240                        #
241                        self.disconnect()
242                        self.connect()
243
244                return self.s.makefile( 'r' )
245
246class GangliaXMLProcessor:
247
248        def __init__( self ):
249                "Setup initial XML connection and handlers"
250
251                self.config = GangliaConfigParser( GMETAD_CONF )
252
253                self.myXMLGatherer = GangliaXMLGatherer( 'localhost', 8651 ) 
254                self.myParser = make_parser()   
255                self.myHandler = GangliaXMLHandler( self.config )
256                self.myParser.setContentHandler( self.myHandler )
257
258        def daemon( self ):
259                "Run as daemon forever"
260
261                self.DAEMON = 1
262
263                # Fork the first child
264                #
265                pid = os.fork()
266
267                if pid > 0:
268
269                        sys.exit(0)  # end parent
270
271                # creates a session and sets the process group ID
272                #
273                os.setsid()
274
275                # Fork the second child
276                #
277                pid = os.fork()
278
279                if pid > 0:
280
281                        sys.exit(0)  # end parent
282
283                # Go to the root directory and set the umask
284                #
285                os.chdir('/')
286                os.umask(0)
287
288                sys.stdin.close()
289                sys.stdout.close()
290                #sys.stderr.close()
291
292                os.open('/dev/null', 0)
293                os.dup(0)
294                os.dup(0)
295
296                self.run()
297
298        def printTime( self ):
299                "Print current time in human readable format"
300
301                return time.strftime("%a %d %b %Y %H:%M:%S")
302
303        def run( self ):
304                "Main thread"
305
306                xmlthread = threading.Thread( None, self.processXML, 'xmlthread' )
307                storethread = threading.Thread( None, self.storeMetrics, 'storethread' )
308
309                while( 1 ):
310
311                        if not xmlthread.isAlive():
312                                # Gather XML at the same interval as gmetad
313
314                                # threaded call to: self.processXML()
315                                #
316                                xmlthread = threading.Thread( None, self.processXML, 'xmlthread' )
317                                xmlthread.start()
318
319                        if not storethread.isAlive():
320                                # Store metrics every .. sec
321
322                                # threaded call to: self.storeMetrics()
323                                #
324                                storethread = threading.Thread( None, self.storeMetrics, 'storethread' )
325                                storethread.start()
326               
327                        # Just sleep a sec here, to prevent daemon from going mad. We're all threads here anyway
328                        time.sleep( 1 ) 
329
330        def storeMetrics( self ):
331                "Store metrics retained in memory to disk"
332
333                debug_msg( 7, self.printTime() + ' - storethread(): started.' )
334
335                # Store metrics somewhere between every 60 and 180 seconds
336                #
337                #STORE_INTERVAL = random.randint( 60, 180 )
338                STORE_INTERVAL = 40
339
340                storethread = threading.Thread( None, self.storeThread, 'storemetricthread' )
341                storethread.start()
342
343                debug_msg( 7, self.printTime() + ' - storethread(): Sleeping.. (%ss)' %STORE_INTERVAL )
344                time.sleep( STORE_INTERVAL )
345                debug_msg( 7, self.printTime() + ' - storethread(): Done sleeping.' )
346
347                if storethread.isAlive():
348
349                        debug_msg( 7, self.printTime() + ' - storethread(): storemetricthread() still running, waiting to finish..' )
350                        storethread.join( 180 ) # Maximum time is for storing thread to finish
351                        debug_msg( 7, self.printTime() + ' - storethread(): Done waiting.' )
352
353                debug_msg( 7, self.printTime() + ' - storethread(): finished.' )
354
355                return 0
356
357        def storeThread( self ):
358
359                debug_msg( 7, self.printTime() + ' - storemetricthread(): started.' )
360                debug_msg( 7, self.printTime() + ' - storemetricthread(): Storing data..' )
361                ret = self.myHandler.storeMetrics()
362                debug_msg( 7, self.printTime() + ' - storemetricthread(): Done storing.' )
363                debug_msg( 7, self.printTime() + ' - storemetricthread(): finished.' )
364               
365                return ret
366
367        def processXML( self ):
368                "Process XML"
369
370                debug_msg( 7, self.printTime() + ' - xmlthread(): started.' )
371
372                parsethread = threading.Thread( None, self.parseThread, 'parsethread' )
373                parsethread.start()
374
375                debug_msg( 7, self.printTime() + ' - xmlthread(): Sleeping.. (%ss)' %self.config.getLowestInterval() )
376                time.sleep( float( self.config.getLowestInterval() ) ) 
377                debug_msg( 7, self.printTime() + ' - xmlthread(): Done sleeping.' )
378
379                if parsethread.isAlive():
380
381                        debug_msg( 7, self.printTime() + ' - xmlthread(): parsethread() still running, waiting to finish..' )
382                        parsethread.join( 60 ) # Maximum time for XML thread to finish
383                        debug_msg( 7, self.printTime() + ' - xmlthread(): Done waiting.' )
384
385                debug_msg( 7, self.printTime() + ' - xmlthread(): finished.' )
386
387                return 0
388
389        def parseThread( self ):
390
391                debug_msg( 7, self.printTime() + ' - parsethread(): started.' )
392                debug_msg( 7, self.printTime() + ' - parsethread(): Parsing XML..' )
393                ret = self.myParser.parse( self.myXMLGatherer.getFileObject() )
394                debug_msg( 7, self.printTime() + ' - parsethread(): Done parsing.' )
395                debug_msg( 7, self.printTime() + ' - parsethread(): finished.' )
396
397                return ret
398
399class GangliaConfigParser:
400
401        sources = [ ]
402
403        def __init__( self, config ):
404
405                self.config = config
406                self.parseValues()
407
408        def parseValues( self ):
409                "Parse certain values from gmetad.conf"
410
411                readcfg = open( self.config, 'r' )
412
413                for line in readcfg.readlines():
414
415                        if line.count( '"' ) > 1:
416
417                                if line.find( 'data_source' ) != -1 and line[0] != '#':
418
419                                        source = { }
420                                        source['name'] = line.split( '"' )[1]
421                                        source_words = line.split( '"' )[2].split( ' ' )
422
423                                        for word in source_words:
424
425                                                valid_interval = 1
426
427                                                for letter in word:
428
429                                                        if letter not in string.digits:
430
431                                                                valid_interval = 0
432
433                                                if valid_interval and len(word) > 0:
434
435                                                        source['interval'] = word
436                                                        debug_msg( 9, 'polling interval for %s = %s' %(source['name'], source['interval'] ) )
437       
438                                        # No interval found, use Ganglia's default     
439                                        if not source.has_key( 'interval' ):
440                                                source['interval'] = 15
441                                                debug_msg( 9, 'polling interval for %s defaulted to 15' %(source['name']) )
442
443                                        self.sources.append( source )
444
445        def getInterval( self, source_name ):
446
447                for source in self.sources:
448
449                        if source['name'] == source_name:
450
451                                return source['interval']
452
453                return None
454
455        def getLowestInterval( self ):
456
457                lowest_interval = 0
458
459                for source in self.sources:
460
461                        if not lowest_interval or source['interval'] <= lowest_interval:
462
463                                lowest_interval = source['interval']
464
465                # Return 15 when nothing is found, so that the daemon won't go insane with 0 sec delays
466                if lowest_interval:
467                        return lowest_interval
468                else:
469                        return 15
470
471class RRDHandler:
472
473        myMetrics = { }
474        lastStored = { }
475        slot = None
476
477        def __init__( self, config, cluster ):
478                self.cluster = cluster
479                self.config = config
480                self.slot = threading.Lock()
481                self.rrdm = RRDMutator()
482
483        def getClusterName( self ):
484                return self.cluster
485
486        def memMetric( self, host, metric ):
487
488                if self.myMetrics.has_key( host ):
489
490                        if self.myMetrics[ host ].has_key( metric['name'] ):
491
492                                for mymetric in self.myMetrics[ host ][ metric['name'] ]:
493
494                                        if mymetric['time'] == metric['time']:
495
496                                                # Allready have this metric, abort
497                                                return 1
498                        else:
499                                self.myMetrics[ host ][ metric['name'] ] = [ ]
500                else:
501                        self.myMetrics[ host ] = { }
502                        self.myMetrics[ host ][ metric['name'] ] = [ ]
503
504                # Ah, psst, push it
505                #
506                # <atomic>
507                self.slot.acquire()
508
509                self.myMetrics[ host ][ metric['name'] ].append( metric )
510
511                self.slot.release()
512                # </atomic>
513
514        def makeUpdateList( self, host, metricname ):
515
516                update_list = [ ]
517
518                while len( self.myMetrics[ host ][ metricname ] ) > 0:
519
520                        # Kabouter pop
521                        #
522                        # <atomic>     
523                        self.slot.acquire()
524
525                        metric = self.myMetrics[ host ][ metricname ].pop()
526
527                        self.slot.release()
528                        # </atomic>
529
530                        if self.checkStoreMetric( host, metricname, metric ):
531                                update_list.append( '%s:%s' %( metric['time'], metric['val'] ) )
532                        else:
533                                print 'allready wrote metric %s with timestamp %s' %( metric['name'], metric['time'] )
534
535                return update_list
536
537        def checkStoreMetric( self, host, metricname, metric ):
538
539                if self.lastStored.has_key( host ):
540
541                        if self.lastStored[ host ].has_key( metricname ):
542
543                                if self.lastStored[ host ][ metricname ] <= metric['time']:
544
545                                        # Allready wrote a value with this timestamp, skip tnx
546                                        return 0
547
548                else:
549                        self.lastStored[ host ] = { }
550
551                self.lastStored[ host ][ metricname ] = metric['time']
552
553                return 1
554
555        def storeMetrics( self ):
556
557                for hostname, mymetrics in self.myMetrics.items():     
558
559                        for metricname, mymetric in mymetrics.items():
560
561                                mytime = self.makeTimeSerial()
562                                correct_serial = self.checkNewRrdPeriod( hostname, mytime )
563                                self.createCheck( hostname, metricname, correct_serial )       
564                                update_ret = self.update( hostname, metricname, correct_serial )
565
566                                if update_ret == 0:
567
568                                        debug_msg( 9, 'stored metric %s for %s' %( hostname, metricname ) )
569                                else:
570                                        debug_msg( 9, 'metric update failed' )
571                                        return 1
572
573                                return 1
574
575        def makeTimeSerial( self ):
576                "Generate a time serial. Seconds since epoch"
577
578                # Seconds since epoch
579                mytime = int( time.time() )
580
581                return mytime
582
583        def makeRrdPath( self, host, metricname=None, timeserial=None ):
584                """
585                Make a RRD location/path and filename
586                If a metric or timeserial are supplied the complete locations
587                will be made, else just the host directory
588                """
589
590                if not timeserial:     
591                        rrd_dir = '%s/%s/%s' %( check_dir(ARCHIVE_PATH), self.cluster, host )
592                else:
593                        rrd_dir = '%s/%s/%s/%s' %( check_dir(ARCHIVE_PATH), self.cluster, host, timeserial )
594                if metricname:
595                        rrd_file = '%s/%s.rrd' %( rrd_dir, metricname )
596                else:
597                        rrd_file = None
598
599                return rrd_dir, rrd_file
600
601        def getLastRrdTimeSerial( self, host ):
602                """
603                Find the last timeserial (directory) for this host
604                This is determined once every host
605                """
606
607                rrd_dir, rrd_file = self.makeRrdPath( host )
608
609                newest_timeserial = 0
610
611                if os.path.exists( rrd_dir ):
612
613                        for root, dirs, files in os.walk( rrd_dir ):
614
615                                for dir in dirs:
616
617                                        valid_dir = 1
618
619                                        for letter in dir:
620                                                if letter not in string.digits:
621                                                        valid_dir = 0
622
623                                        if valid_dir:
624                                                timeserial = dir
625                                                if timeserial > newest_timeserial:
626                                                        newest_timeserial = timeserial
627
628                if newest_timeserial:
629                        return newest_timeserial
630                else:
631                        return 0
632
633        def checkNewRrdPeriod( self, host, current_timeserial ):
634                """
635                Check if current timeserial belongs to recent time period
636                or should become a new period (and file).
637
638                Returns the serial of the correct time period
639                """
640
641                last_timeserial = int( self.getLastRrdTimeSerial( host ) )
642                debug_msg( 9, 'last timeserial of %s is %s' %( host, last_timeserial ) )
643
644                if not last_timeserial:
645                        serial = current_timeserial
646                else:
647
648                        archive_secs = ARCHIVE_HOURS_PER_RRD * (60 * 60)
649
650                        if (current_timeserial - last_timeserial) >= archive_secs:
651                                serial = current_timeserial
652                        else:
653                                serial = last_timeserial
654
655                return serial
656
657        def getFirstTime( self, host, metricname ):
658                "Get the first time of a metric we know of"
659
660                first_time = 0
661
662                for metric in self.myMetrics[ host ][ metricname ]:
663
664                        if not first_time or metric['time'] <= first_time:
665
666                                first_time = metric['time']
667
668                return first_time
669
670        def createCheck( self, host, metricname, timeserial ):
671                "Check if an .rrd allready exists for this metric, create if not"
672
673                debug_msg( 9, 'rrdcreate: using timeserial %s for %s/%s' %( timeserial, host, metricname ) )
674
675                rrd_dir, rrd_file = self.makeRrdPath( host, metricname, timeserial )
676
677                if not os.path.exists( rrd_dir ):
678                        os.makedirs( rrd_dir )
679                        debug_msg( 9, 'created dir %s' %( str(rrd_dir) ) )
680
681                if not os.path.exists( rrd_file ):
682
683                        interval = self.config.getInterval( self.cluster )
684                        heartbeat = 8 * int(interval)
685
686                        params = [ ]
687
688                        params.append( '--step' )
689                        params.append( str( interval ) )
690
691                        params.append( '--start' )
692                        params.append( str( int( self.getFirstTime( host, metricname ) ) - 1 ) )
693
694                        params.append( 'DS:sum:GAUGE:%d:U:U' %heartbeat )
695                        params.append( 'RRA:AVERAGE:0.5:1:%s' %(ARCHIVE_HOURS_PER_RRD * 240) )
696
697                        self.rrdm.create( str(rrd_file), params )
698
699                        debug_msg( 9, 'created rrd %s' %( str(rrd_file) ) )
700
701        def update( self, host, metricname, timeserial ):
702
703                debug_msg( 9, 'rrdupdate: using timeserial %s for %s/%s' %( timeserial, host, metricname ) )
704
705                rrd_dir, rrd_file = self.makeRrdPath( host, metricname, timeserial )
706
707                update_list = self.makeUpdateList( host, metricname )
708
709                ret = self.rrdm.update( str(rrd_file), update_list )
710
711                if ret:
712                        return 1
713               
714                debug_msg( 9, 'updated rrd %s with %s' %( str(rrd_file), string.join( update_list ) ) )
715
716                return 0
717
718def main():
719        "Program startup"
720
721        myProcessor = GangliaXMLProcessor()
722
723        if DAEMONIZE:
724                myProcessor.daemon()
725        else:
726                myProcessor.run()
727
728def check_dir( directory ):
729        "Check if directory is a proper directory. I.e.: Does _not_ end with a '/'"
730
731        if directory[-1] == '/':
732                directory = directory[:-1]
733
734        return directory
735
736def debug_msg( level, msg ):
737
738        if (DEBUG_LEVEL >= level):
739                sys.stderr.write( msg + '\n' )
740
741# Let's go
742if __name__ == '__main__':
743        main()
Note: See TracBrowser for help on using the repository browser.