Changeset 774

Timestamp:

03/29/13 14:07:28 (11 years ago)

Author:

ramonb

Message:

• job_id is now varchar and not int anymore (due to array ids)
• fixed monarch job detection: jobid is now last field and sequence number is first

File:

• branches/0.4/jobarchived/jobarchived.py (modified) (17 diffs)

branches/0.4/jobarchived/jobarchived.py

@@ -134 +134 @@
     global DEBUG_LEVEL, USE_SYSLOG, SYSLOG_LEVEL, SYSLOG_FACILITY, GMETAD_CONF, ARCHIVE_XMLSOURCE
     global ARCHIVE_DATASOURCES, ARCHIVE_PATH, ARCHIVE_HOURS_PER_RRD, ARCHIVE_EXCLUDE_METRICS
-    global JOB_SQL_DBASE, DAEMONIZE, RRDTOOL, JOB_TIMEOUT, MODRRDTOOL
+    global JOB_SQL_DBASE, DAEMONIZE, RRDTOOL, JOB_TIMEOUT, MODRRDTOOL, JOB_SQL_PASSWORD, JOB_SQL_USER
 
     ARCHIVE_PATH        = cfg.get( 'DEFAULT', 'ARCHIVE_PATH' )
@@ -180 +180 @@
 
     JOB_SQL_DBASE        = cfg.get( 'DEFAULT', 'JOB_SQL_DBASE' )
+    JOB_SQL_USER        = cfg.get( 'DEFAULT', 'JOB_SQL_USER' )
+    JOB_SQL_PASSWORD        = cfg.get( 'DEFAULT', 'JOB_SQL_PASSWORD' )
 
     JOB_TIMEOUT        = cfg.getint( 'DEFAULT', 'JOB_TIMEOUT' )
@@ -328 +330 @@
         try:
             c.execute(q_str)
-            result = c.oidValue
 
         except psycopg2.Error, details:
@@ -336 +337 @@
 
         c.close()
-        return result
+        return True
 
     def Commit(self):
@@ -348 +349 @@
     def __init__( self, hostname, database ):
 
+        global JOB_SQL_USER, JOB_SQL_PASSWORD
+
         self.db_vars = InitVars(DataBaseName=database,
-                User='root',
+                User=JOB_SQL_USER,
                 Host=hostname,
-                Password='',
+                Password=JOB_SQL_PASSWORD,
                 Dictionary='true')
 
@@ -508 +511 @@
         elif action == 'update':
 
-            self.setDatabase( "UPDATE jobs SET %s WHERE job_id=%s" %(update_str, job_id) )
+            self.setDatabase( "UPDATE jobs SET %s WHERE job_id='%s'" %(update_str, job_id) )
 
         if len( ids ) > 0:
@@ -540 +543 @@
     def addJobNode( self, jobid, nodeid ):
 
-        self.setDatabase( "INSERT INTO job_nodes (job_id,node_id) VALUES ( %s,%s )" %(jobid, nodeid) )
+        self.setDatabase( "INSERT INTO job_nodes (job_id,node_id) VALUES ( '%s',%s )" %(jobid, nodeid) )
 
     def storeJobInfo( self, jobid, jobattrs ):
@@ -790 +793 @@
 
             my_data    = self.myXMLSource.getData()
+            #print my_data
+            #print "size my data: %d" %len( my_data )
 
             debug_msg( 1, 'torque_xml_thread(): Done retrieving.' )
@@ -799 +804 @@
 
                 debug_msg( 1, 'ganglia_parse_thread(): Done parsing.' )
+            else:
+                debug_msg( 1, 'torque_xml_thread(): Got no data.' )
+
                 
             debug_msg( 1, 'torque_xml_thread(): Sleeping.. (%ss)' %(str( self.config.getLowestInterval() ) ) )
@@ -814 +822 @@
         self.jobs_processed    = [ ]
         self.jobs_to_store    = [ ]
+        debug_msg( 1, "XML: Handler created" )
 
     def startDocument( self ):
@@ -819 +828 @@
         self.heartbeat    = 0
         self.elementct    = 0
+        debug_msg( 1, "XML: Start document" )
 
     def startElement( self, name, attrs ):
@@ -844 +854 @@
             elif metricname.find( 'zplugin_monarch_job' ) != -1:
 
-                job_id    = metricname.split( 'zplugin_monarch_job_' )[1].split( '_' )[0]
+                job_id    = metricname.split( 'zplugin_monarch_job_' )[1].split( '_' )[1]
                 val    = str( attrs.get( 'VAL', "" ) )
 
@@ -1331 +1341 @@
         # Store metrics somewhere between every 360 and 640 seconds
         #
-        if DEBUG_LEVEL > 2:
-            #STORE_INTERVAL = 60
-            STORE_INTERVAL = random.randint( 360, 640 )
+        if DEBUG_LEVEL >= 1:
+            STORE_INTERVAL = 60
+            #STORE_INTERVAL = random.randint( 360, 640 )
         else:
             STORE_INTERVAL = random.randint( 360, 640 )
@@ -1352 +1362 @@
         if store_metric_thread.isAlive():
 
-            debug_msg( 1, 'ganglia_store_thread(): storemetricthread() still running, waiting to finish..' )
+            debug_msg( 1, 'ganglia_store_thread(): storemetricthread() still running, waiting to terminate..' )
             store_metric_thread.join( STORE_TIMEOUT ) # Maximum time is for storing thread to finish
-            debug_msg( 1, 'ganglia_store_thread(): Done waiting.' )
+            debug_msg( 1, 'ganglia_store_thread(): Done waiting: terminated storemetricthread()' )
 
         debug_msg( 1, 'ganglia_store_thread(): finished.' )
@@ -1376 +1386 @@
         """Process XML"""
 
+        debug_msg( 5, "Entering processXML()")
+
         try:
             parsethread = threading.Thread( None, self.parseThread, 'parsethread' )
@@ -1391 +1403 @@
         if parsethread.isAlive():
 
-            debug_msg( 1, 'ganglia_xml_thread(): parsethread() still running, waiting (%ss) to finish..' %PARSE_TIMEOUT )
+            debug_msg( 1, 'ganglia_xml_thread(): parsethread() still running, waiting (%ss) to terminate..' %PARSE_TIMEOUT )
             parsethread.join( PARSE_TIMEOUT ) # Maximum time for XML thread to finish
-            debug_msg( 1, 'ganglia_xml_thread(): Done waiting.' )
+            debug_msg( 1, 'ganglia_xml_thread(): Done waiting. parsethread() terminated' )
 
         debug_msg( 1, 'ganglia_xml_thread(): finished.' )
+
+        debug_msg( 5, "Leaving processXML()")
 
         return 0
@@ -1406 +1420 @@
         
         my_data    = self.myXMLSource.getData()
+        debug_msg( 1, 'ganglia_parse_thread(): data size %d.' %len(my_data) )
 
         debug_msg( 1, 'ganglia_parse_thread(): Done retrieving.' )