Changeset 365 for trunk/jobarchived

Timestamp:

06/13/07 10:20:05 (16 years ago)

Author:

bastiaans

Message:

jobarchived/jobarchived.py:

• rewritten to use py-rrdtool api and no more pipes

File:

• trunk/jobarchived/jobarchived.py (modified) (19 diffs)

trunk/jobarchived/jobarchived.py

@@ -163 +163 @@
 import DBClass
 import xml.sax, xml.sax.handler, socket, string, os, os.path, time, thread, threading, random, re
+import rrdtool
 
 class DataSQLStore:
@@ -193 +194 @@
         def doDatabase(self, type, statement):
 
-                debug_msg( 6, 'doDatabase(): %s: %s' %(type, statement) )
+                debug_msg( 10, 'doDatabase(): %s: %s' %(type, statement) )
                 try:
                         if type == 'set':
@@ -206 +207 @@
                         sys.exit(1)
 
-                debug_msg( 6, 'doDatabase(): result: %s' %(result) )
+                debug_msg( 10, 'doDatabase(): result: %s' %(result) )
                 return result
 
@@ -272 +273 @@
                 update_str      = None
 
-                debug_msg( 6, 'mutateJob(): %s %s' %(action,job_id))
+                debug_msg( 10, 'mutateJob(): %s %s' %(action,job_id))
 
                 ids = [ ]
@@ -434 +435 @@
         binary = None
 
+
         def __init__( self, binary=None ):
                 """Set alternate binary if supplied"""
@@ -443 +445 @@
                 """Create a new rrd with args"""
 
-                return self.perform( 'create', '"' + filename + '"', args )
+                #return self.perform( 'create', '"' + filename + '"', args )
+                return self.perform( 'create', filename, args )
 
         def update( self, filename, args ):
                 """Update a rrd with args"""
 
-                return self.perform( 'update', '"' + filename + '"', args )
+                #return self.perform( 'update', '"' + filename + '"', args )
+                return self.perform( 'update', filename, args )
 
         def grabLastUpdate( self, filename ):
@@ -455 +459 @@
                 last_update = 0
 
-                debug_msg( 8, self.binary + ' info "' + filename + '"' )
-
-                my_pipe         = os.popen( self.binary + ' info "' + filename + '"' )
-
-                for line in my_pipe.readlines():
-
-                        if line.find( 'last_update') != -1:
-
-                                last_update = line.split( ' = ' )[1]
-
-                if my_pipe:
-
-                        my_pipe.close()
-
-                if last_update:
+                debug_msg( 8, 'rrdtool.info( ' + filename + ' )' )
+
+                rrd_header      = { }
+
+                try:
+                        rrd_header      = rrdtool.info( filename )
+                except rrdtool.error, msg:
+                        debug_msg( 8, str( msg ) )
+
+                if rrd_header.has_key( 'last_update' ):
                         return last_update
                 else:
@@ -491 +490 @@
                                 arg_string = arg_string + ' ' + arg
 
-                debug_msg( 8, self.binary + ' ' + action + ' ' + filename + ' ' + arg_string  )
-
-                cmd     = os.popen( self.binary + ' ' + action + ' ' + filename + ' ' + arg_string )
-                lines   = cmd.readlines()
-
-                cmd.close()
-
-                for line in lines:
-
-                        if line.find( 'ERROR' ) != -1:
-
-                                error_msg = string.join( line.split( ' ' )[1:] )
-                                debug_msg( 8, error_msg )
-                                return 1
+                debug_msg( 8, 'rrdtool.' + action + "( " + filename + ' ' + arg_string + ")" )
+
+                try:
+                        debug_msg( 8, "filename '" + str(filename) + "' type "+ str(type(filename)) + " args " + str( args ) )
+
+                        if action == 'create':
+
+                                rrdtool.create( str( filename ), *args )
+
+                        elif action == 'update':
+
+                                rrdtool.update( str( filename ), *args )
+
+                except rrdtool.error, msg:
+
+                        error_msg = str( msg )
+                        debug_msg( 8, error_msg )
+                        return 1
 
                 return 0
@@ -566 +569 @@
 
                 self.heartbeat  = 0
+                self.elementct  = 0
 
         def startElement( self, name, attrs ):
@@ -576 +580 @@
                 jobinfo = { }
 
+                self.elementct  += 1
+
                 if name == 'CLUSTER':
 
@@ -623 +629 @@
                                                         self.jobs_to_store.append( job_id )
 
-                                                debug_msg( 6, 'jobinfo for job %s has changed' %job_id )
+                                                debug_msg( 10, 'jobinfo for job %s has changed' %job_id )
                                 else:
                                         self.jobAttrs[ job_id ] = jobinfo
@@ -630 +636 @@
                                                 self.jobs_to_store.append( job_id )
 
-                                        debug_msg( 6, 'jobinfo for job %s has changed' %job_id )
+                                        debug_msg( 10, 'jobinfo for job %s has changed' %job_id )
                                         
         def endDocument( self ):
                 """When all metrics have gone, check if any jobs have finished"""
+
+                debug_msg( 1, "XML: Processed "+str(self.elementct)+ " elements - found "+str(len(self.jobs_to_store))+" jobs" )
 
                 if self.heartbeat:
@@ -719 +727 @@
 
                 debug_msg( 1, 'Checking database..' )
-                self.ds.checkStaleJobs()
+
+                global DEBUG_LEVEL
+
+                if DEBUG_LEVEL <= 2:
+                        self.ds.checkStaleJobs()
+
                 debug_msg( 1, 'Check done.' )
                 debug_msg( 1, 'Checking rrd archive..' )
@@ -743 +756 @@
 
                                         self.clusters[ clustername ] = RRDHandler( self.config, clustername )
+
+                debug_msg( 9, "Found "+str(len(self.clusters.keys()))+" clusters" )
 
         def startElement( self, name, attrs ):
@@ -1048 +1063 @@
                 """Store metrics retained in memory to disk"""
 
+                global DEBUG_LEVEL
+
                 # Store metrics somewhere between every 360 and 640 seconds
                 #
-                STORE_INTERVAL = random.randint( 360, 640 )
+                if DEBUG_LEVEL > 2:
+                        #STORE_INTERVAL = 60
+                        STORE_INTERVAL = random.randint( 360, 640 )
+                else:
+                        STORE_INTERVAL = random.randint( 360, 640 )
 
                 try:
@@ -1227 +1248 @@
                 self.rrdm       = RRDMutator( RRDTOOL )
 
-                self.gatherLastUpdates()
+                global DEBUG_LEVEL
+
+                if DEBUG_LEVEL <= 2:
+                        self.gatherLastUpdates()
 
         def gatherLastUpdates( self ):
@@ -1330 +1354 @@
                         if self.checkStoreMetric( host, metric ):
 
-                                update_list.append( '%s:%s' %( metric['time'], metric['val'] ) )
+                                u_val   = str( metric['time'] ) + ':' + str( metric['val'] )
+                                #update_list.append( str('%s:%s') %( metric['time'], metric['val'] ) )
+                                update_list.append( u_val )
 
                 return update_list
@@ -1380 +1406 @@
                 """
 
+                debug_msg( 5, "Entering storeMetrics()")
+
+                count_values    = 0
+                count_metrics   = 0
+                count_bits      = 0
+
+                for hostname, mymetrics in self.myMetrics.items():      
+
+                        for metricname, mymetric in mymetrics.items():
+
+                                count_metrics += 1
+
+                                for dmetric in mymetric:
+
+                                        count_values += 1
+
+                                        count_bits      += len( dmetric['time'] )
+                                        count_bits      += len( dmetric['val'] )
+
+                count_bytes     = count_bits / 8
+
+                debug_msg( 5, "size of cluster '" + self.cluster + "': " + 
+                        str( len( self.myMetrics.keys() ) ) + " hosts " + 
+                        str( count_metrics ) + " metrics " + str( count_values ) + " values " +
+                        str( count_bits ) + " bits " + str( count_bytes ) + " bytes " )
+
                 for hostname, mymetrics in self.myMetrics.items():      
 
@@ -1437 +1489 @@
 
                                 self.memLastUpdate( hostname, metricname, metrics_to_store )
+
+                debug_msg( 5, "Leaving storeMetrics()")
 
         def makeTimeSerial( self ):