source: trunk/daemon/togad.py @ 10

#!/usr/bin/env python

from xml.sax import make_parser
from xml.sax.handler import ContentHandler 
import socket
import sys
import rrdtool
import string

# Specify debugging level here;
#
# >10 = metric XML
# >9  = host,cluster,grid,ganglia XML
# >8  = RRD activity,gmetad config parsing
#
DEBUG_LEVEL = 9

# Where is the gmetad.conf located
#
GMETAD_CONF = '/etc/gmetad.conf'

# Where to store the archived rrd's
#
ARCHIVE_PATH = '/data/toga/rrds'

# List of data_source names to archive for
#
ARCHIVE_SOURCES = [ "LISA Cluster" ]

#
# Configuration ends
#

"""
This is TOrque-GAnglia's data Daemon
"""

class GangliaXMLHandler( ContentHandler ):
        "Parse Ganglia's XML"

        metrics = [ ]

        def startElement( self, name, attrs ):
                "Store appropriate data from xml start tags"

                if name == 'GANGLIA_XML':
                        self.XMLSource = attrs.get('SOURCE',"")
                        self.gangliaVersion = attrs.get('VERSION',"")
                        if (DEBUG_LEVEL>9): print 'Found XML data: source %s version %s' %( self.XMLSource, self.gangliaVersion )

                elif name == 'GRID':
                        self.gridName = attrs.get('NAME',"")
                        if (DEBUG_LEVEL>9): print '`-Grid found: %s' %( self.gridName )

                elif name == 'CLUSTER':
                        self.clusterName = attrs.get('NAME',"")
                        self.rrd = RRDHandler( self.clusterName )
                        if (DEBUG_LEVEL>9): print ' |-Cluster found: %s' %( self.clusterName )

                elif name == 'HOST':     
                        self.hostName = attrs.get('NAME',"")
                        self.hostIp = attrs.get('IP',"")
                        self.hostReported = attrs.get('REPORTED',"")
                        # Reset the metrics list for each host
                        self.metrics = [ ]
                        if (DEBUG_LEVEL>9): print ' | |-Host found: %s - ip %s reported %s' %( self.hostName, self.hostIp, self.hostReported )

                elif name == 'METRIC':
                        myMetric = { }
                        myMetric['name'] = attrs.get('NAME',"")
                        myMetric['val'] = attrs.get('VAL',"")

                        self.metrics.append( myMetric ) 
                        if (DEBUG_LEVEL>10): print ' | | |-metric: %s:%s' %( myMetric['name'], myMetric['val'] )

                return

        def endElement( self, name ):
                #if name == 'ganglia_xml':

                #if name == 'grid':

                #if name == 'cluster':

                if name == 'host':     
                        self.storeMetrics( self.hostName )

                #if name == 'metric':

        def storeMetrics( self, hostname ):

                for metric in self.metrics:
                        self.rrd.checkCreate( hostname, metric['name'] )        
                        self.rrd.update( hostname, metric['name'], metric['val'] )
                        if (DEBUG_LEVEL>8): print 'stored metric %s for %s: %s' %( hostname, metric['name'], metric['val'] )
                        sys.exit(1)
        

class GangliaXMLGatherer:
        "Setup a connection and file object to Ganglia's XML"

        s = None

        def __init__( self, host, port ):
                "Store host and port for connection"

                self.host = host
                self.port = port

        def __del__( self ):
                "Kill the socket before we leave"

                self.s.close()

        def getFileObject( self ):
                "Connect, and return a file object"

                for res in socket.getaddrinfo( self.host, self.port, socket.AF_UNSPEC, socket.SOCK_STREAM ):
                        af, socktype, proto, canonname, sa = res
                        try:
                                self.s = socket.socket( af, socktype, proto )
                        except socket.error, msg:
                                self.s = None
                                continue
                        try:
                                self.s.connect( sa )
                        except socket.error, msg:
                                self.s.close()
                                self.s = None
                                continue
                        break

                if self.s is None:
                        print 'Could not open socket'
                        sys.exit(1)

                return self.s.makefile( 'r' )

class GangliaXMLProcessor:

        def daemon( self ):
                "Run as daemon forever"

                self.DAEMON = 1

                # Fork the first child
                #
                pid = os.fork()
                if pid > 0:
                        sys.exit(0)  # end parrent

                # creates a session and sets the process group ID 
                #
                os.setsid()

                # Fork the second child
                #
                pid = os.fork()
                if pid > 0:
                        sys.exit(0)  # end parrent

                # Go to the root directory and set the umask
                #
                os.chdir('/')
                os.umask(0)

                sys.stdin.close()
                sys.stdout.close()
                sys.stderr.close()

                os.open('/dev/null', 0)
                os.dup(0)
                os.dup(0)

                self.run()

        def run( self ):
                "Main thread"

                while ( 1 ):
                        self.processXML()
                        time.sleep( 5 )

        def processXML( self ):
                "Process XML"

                myXMLGatherer = GangliaXMLGatherer( 'localhost', 8651 ) 

                myParser = make_parser()   
                myHandler = GangliaXMLHandler()
                myParser.setContentHandler( myHandler )

                myParser.parse( myXMLGatherer.getFileObject() )

class GangliaConfigParser:

        sources = [ ]

        def __init__( self, config ):
                self.config = config
                self.parseValues()

        def parseValues(self):
                "Parse certain values from gmetad.conf"

                readcfg = open( self.config, 'r' )

                for line in readcfg.readlines():

                        if line.count( '"' ) > 1:

                                source = { }
                                source['name'] = line.split( '"' )[1]

                                if line.find( 'data_source' ) != -1 and line[0] != '#':

                                        source_words = line.split( '"' )[2].split( ' ' )

                                        for word in source_words:

                                                valid_interval = 1

                                                for letter in word:
                                                        if letter not in string.digits:
                                                                valid_interval = 0

                                                if valid_interval and len(word) > 0:
                                                        source['interval'] = word
                                                        if (DEBUG_LEVEL>8): print 'polling interval for %s = %s' %(source['name'], source['interval'] )
                
                # No interval found, use Ganglia's default      
                if not source.has_key( 'interval' ):
                        source['interval'] = 15

                self.sources.append( source )

        def getInterval( self, source_name ):
                for source in self.sources:
                        if source['name'] == name:
                                return source['interval']
                return None

class RRDHandler:

        def __init__( self, cluster ):
                self.cluster = cluster

                self.gmetad_conf = GangliaConfigParser( GMETAD_CONF )

        def createCheck( self, host, metric ):
                "Check if an .rrd allready exists for this metric, create if not"

                rrd_dir = '%s/%s/%s' %( check_dir(ARCHIVE_PATH), self.cluster, host )

                if not os.path.exists( rrd_dir ):
                        os.makedirs( rrd_dir )

                rrd_file = '%s/%s.rrd' %( rrd_dir, metric )

                interval = self.gmetad_conf.getInterval( self.cluster )

        def update( self, metric, timestamp, val ):

                pass

                #rrd.update( bla )
                

def main():
        "Program startup"

        myProcessor = GangliaXMLProcessor()
        myProcessor.processXML()

def check_dir( directory ):
        "Check if directory is a proper directory. I.e.: Does _not_ end with a '/'"

        if directory[-1] == '/':
                directory = directory[:-1]

        return directory

# Let's go
if __name__ == '__main__':
        main()