Changeset 54

Timestamp:

04/08/05 12:11:16 (18 years ago)

Author:

bastiaans

Message:

daemon/togad.py:

• Several bugfixes
• Now more than 1 storethread is possible

File:

• trunk/daemon/togad.py (modified) (11 diffs)

trunk/daemon/togad.py

@@ -23 +23 @@
 # 7  = daemon threading
 #
-DEBUG_LEVEL = 7
+DEBUG_LEVEL = 8
 
 # Where is the gmetad.conf located
@@ -63 +63 @@
 STORE_TIMEOUT = 360
 
+# Number of storing threads
+#
+STORE_THREADS = 1
 """
 This is TOrque-GAnglia's data Daemon
@@ -116 +119 @@
                                 arg_string = arg_string + ' ' + arg
 
-                debug_msg( 8, 'rrdm.perform(): ' + self.binary + ' ' + action + ' ' + filename + ' ' + arg_string  )
+                debug_msg( 8, self.binary + ' ' + action + ' ' + filename + ' ' + arg_string  )
 
                 for line in os.popen( self.binary + ' ' + action + ' ' + filename + ' ' + arg_string ).readlines():
@@ -134 +137 @@
                 self.config = config
                 self.clusters = { }
-                debug_msg( 0, printTime() + ' Checking existing toga rrd archive..' )
+                debug_msg( 0, printTime() + ' - Checking existing toga rrd archive..' )
                 self.gatherClusters()
-                debug_msg( 0, printTime() + ' Check done.' )
+                debug_msg( 0, printTime() + ' - Check done.' )
 
         def gatherClusters( self ):
@@ -349 +352 @@
 
                 xmlthread = threading.Thread( None, self.processXML, 'xmlthread' )
-                storethread = threading.Thread( None, self.storeMetrics, 'storethread' )
+
+                storethreads = [ ]
+
+                for num in range( int( STORE_THREADS ) ):
+
+                        storethreads.append( threading.Thread( None, self.storeMetrics, 'storethread' ) )
 
                 while( 1 ):
@@ -361 +369 @@
                                 xmlthread.start()
 
-                        if not storethread.isAlive():
-                                # Store metrics every .. sec
-
-                                # threaded call to: self.storeMetrics()
-                                #
-                                storethread = threading.Thread( None, self.storeMetrics, 'storethread' )
-                                storethread.start()
+                        for storethread in storethreads:
+
+                                mythread = storethreads.index( storethread )
+
+                                if not storethread.isAlive():
+
+                                        # Store metrics every .. sec
+
+                                        # threaded call to: self.storeMetrics()
+                                        #
+                                        storethreads[ mythread ] = threading.Thread( None, self.storeMetrics, 'storethread' )
+                                        storethreads[ mythread ].start()
                 
                         # Just sleep a sec here, to prevent daemon from going mad. We're all threads here anyway
@@ -379 +392 @@
                 # Store metrics somewhere between every 60 and 180 seconds
                 #
-                STORE_INTERVAL = random.randint( 360, 640 )
+                #STORE_INTERVAL = random.randint( 360, 640 )
+                STORE_INTERVAL = 60
 
                 storethread = threading.Thread( None, self.storeThread, 'storemetricthread' )
@@ -631 +645 @@
                 return 1
 
-        def memLastUpdate( self, host, metriclist ):
-
-                last_update_time = 0
-
-                for metric in metriclist:
-
-                        if metric['time'] > last_update_time:
-
-                                last_update_time = metric['time']
+        def memLastUpdate( self, host, metricname, metriclist ):
 
                 if not self.lastStored.has_key( host ):
                         self.lastStored[ host ] = { }
 
-                if self.lastStored[ host ].has_key( metric['name'] ):
+                last_update_time = 0
+
+                for metric in metriclist:
+
+                        if metric['name'] == metricname:
+
+                                if metric['time'] > last_update_time:
+
+                                        last_update_time = metric['time']
+
+                if self.lastStored[ host ].has_key( metricname ):
                         
-                        if last_update_time <= self.lastStored[ host ][ metric['name'] ]:
+                        if last_update_time <= self.lastStored[ host ][ metricname ]:
                                 return 1
 
-                self.lastStored[ host ][ metric['name'] ] = last_update_time
+                self.lastStored[ host ][ metricname ] = last_update_time
 
         def storeMetrics( self ):
@@ -663 +679 @@
                                 self.slot.acquire() 
 
-                                while len( mymetric ) > 0:
-
-                                        metrics_to_store.append( mymetric.pop( 0 ) )
+                                while len( self.myMetrics[ hostname ][ metricname ] ) > 0:
+
+                                        if len( self.myMetrics[ hostname ][ metricname ] ) > 0:
+                                                metrics_to_store.append( self.myMetrics[ hostname ][ metricname ].pop( 0 ) )
 
                                 self.slot.release()
@@ -674 +691 @@
                                 #
                                 metric_serial_table = self.determineSerials( hostname, metricname, metrics_to_store )
-                                self.myMetrics[ hostname ][ metricname ] = [ ]
 
                                 update_rets = [ ]
@@ -694 +710 @@
                                 if not (1) in update_rets:
 
-                                        self.memLastUpdate( hostname, mymetric )
+                                        self.memLastUpdate( hostname, metricname, metrics_to_store )
 
         def makeTimeSerial( self ):