source: trunk/daemon/togad.py @ 42

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

daemon/togad.py:

Added functions for gathering all 'last_update' values of existing .rrds

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