+ new XMLRPC_SERVER name to boot.planet-lab.org
[monitor.git] / rt.py
diff --git a/rt.py b/rt.py
index 4e56869..8cd1278 100644 (file)
--- a/rt.py
+++ b/rt.py
@@ -6,8 +6,11 @@ import string
 import logging
 import Queue
 import time 
+import re
 import comon
+import soltesz
 from threading import *
+import config
 
 # RT database access constants file
 RT_DB_CONSTANTS_PATH='/etc/planetlab/rt_db'
@@ -76,8 +79,8 @@ def open_rt_db():
                                user=rt_db_constants['RT_DB_USER'],
                                passwd=rt_db_constants['RT_DB_PASSWORD'],
                                db=rt_db_constants['RT_DB_NAME'])
-       except Error:
-               print "Failed to connect to RT database"
+       except Exception, err:
+               print "Failed to connect to RT database: %s" %err
                return -1
 
        return rt_db
@@ -85,18 +88,34 @@ def open_rt_db():
 
 
 
-def rt_tickets(hostname):
+def rt_tickets():
        db = open_rt_db()
-       sql = """SELECT distinct Tk.id, Tk.Status, Tk.Subject
-                        FROM Tickets AS Tk
-                        JOIN Transactions AS Tr ON Tk.id=Tr.ObjectId
-                        JOIN Attachments AS At ON Tr.id=At.TransactionID
-                        WHERE (At.Content LIKE '%%%s%%' OR
-                               At.Subject LIKE '%%%s%%') AND
-                               (Tk.Status = 'new' OR Tk.Status = 'open') AND
-                               Tk.Queue = 3
-                        ORDER BY Tk.Status, Tk.LastUpdated DESC""" \
-                        % (hostname,hostname)
+#      sql = """SELECT distinct Tk.id, Tk.Status, Tk.Subject
+#                       FROM Tickets AS Tk
+#                       JOIN Transactions AS Tr ON Tk.id=Tr.ObjectId
+#                       JOIN Attachments AS At ON Tr.id=At.TransactionID
+#                       WHERE (At.Content LIKE '%%%s%%' OR
+#                              At.Subject LIKE '%%%s%%') AND
+#                              (Tk.Status = 'new' OR Tk.Status = 'open') AND
+#                              Tk.Queue = 3 OR Tk.Queue = 19 
+#                       ORDER BY Tk.Status, Tk.LastUpdated DESC""" \
+#                       % (hostname,hostname)
+#      sql = """SELECT distinct Tk.id, Tk.Status, Tk.Subject
+#                       FROM Tickets AS Tk
+#                       JOIN Transactions AS Tr ON Tk.id=Tr.ObjectId
+#                       JOIN Attachments AS At ON Tr.id=At.TransactionID
+#                       WHERE (At.Content LIKE '%%%s%%' OR
+#                              At.Subject LIKE '%%%s%%') AND
+#                              (Tk.Status = 'new' OR Tk.Status = 'open')
+#                       ORDER BY Tk.Status, Tk.LastUpdated DESC""" \
+#                       % (hostname,hostname)
+
+       # Queue == 10 is the spam Queue in RT.
+       sql = """SELECT distinct Tk.id, Tk.Status, Tk.Subject, At.Content
+                        FROM Tickets AS Tk, Attachments AS At 
+                        JOIN Transactions AS Tr ON Tk.id=Tr.ObjectId  
+                        WHERE Tk.Queue != 10 AND Tk.id > 10000 AND 
+                                  Tr.id=At.TransactionID AND (Tk.Status = 'new' OR Tk.Status = 'open')"""
 
        try:
                # create a 'cursor' (required by MySQLdb)
@@ -112,14 +131,51 @@ def rt_tickets(hostname):
        # map list of lists (raw) to list of dicts (tickets) 
        # when int gets pulls from SQL into python ints are converted to LONG to
        # prevent overflow .. convert back
-       tickets = map(lambda x: {"ticket_id":int(x[0]),
+       #tickets = map(lambda x: {"ticket_id":int(x[0]),
+       tickets = map(lambda x: {"ticket_id":str(x[0]),
                                "status":x[1],
-                               "subj":x[2]},
+                               "subj":str(x[2]),
+                               "content":str(x[3])},
                                raw)
        db.close()
 
        return tickets
 
+def is_host_in_rt_tickets(host, ticket_blacklist, ad_rt_tickets):
+       # ad_rt_tickets is an array of dicts, defined above.
+       if len(ad_rt_tickets) == 0:
+               return (False, None)
+       
+       d_ticket = ad_rt_tickets[0]
+       if not ('ticket_id' in d_ticket and 'status' in d_ticket and 
+                       'subj' in d_ticket and 'content' in d_ticket):
+               logger.debug("RT_tickets array has wrong fields!!!")
+               return (False, None)
+
+       #logger.debug("Searching all tickets for %s" % host)
+       def search_tickets(host, ad_rt_tickets):
+               # compile once for more efficiency
+               re_host = re.compile(host)
+               for x in ad_rt_tickets:
+                       if re_host.search(x['subj'], re.MULTILINE|re.IGNORECASE) or \
+                          re_host.search(x['content'], re.MULTILINE|re.IGNORECASE):
+                               logger.debug("\t ticket %s has %s" % (x['ticket_id'], host))
+                               print x['ticket_id']
+                               print ticket_blacklist
+                               if x['ticket_id'] in ticket_blacklist:
+                                       return (False, x)
+                               else:
+                                       return (True, x)
+               logger.debug("\t noticket -- has %s" % host)
+               return (False, None)
+
+       # This search, while O(tickets), takes less than a millisecond, 05-25-07
+       #t = soltesz.MyTimer()
+       ret = search_tickets(host, ad_rt_tickets)
+       #del t
+
+       return ret
+
 
 '''
 Finds tickets associated with hostnames.
@@ -136,27 +192,49 @@ Remove nodes that have come backup. Don't care of ticket is closed after first q
 Another thread refresh tickets of nodes already in dict and remove nodes that have come up. 
 '''
 class RT(Thread):
-       def __init__(self, tickets, bucket, target = None): 
+       def __init__(self, dbTickets, tickets, qin_toCheck, qout_sickNoTicket, l_ticket_blacklist, target = None): 
                # Time of last update of ticket DB
+               self.dbTickets = dbTickets
                self.lastupdated = 0
-               # Queue() is MP/MC self locking 
-               self.bucket = bucket 
+               self.l_ticket_blacklist = l_ticket_blacklist
+               # Check host in queue.  Queue populated from comon data of sick. 
+               self.qin_toCheck = qin_toCheck
+               # Result of rt db query.  Nodes without tickets that are sick.
+               self.qout_sickNoTicket = qout_sickNoTicket 
                #DB of tickets.  Name -> ticket
                self.tickets = tickets
                Thread.__init__(self,target = self.getTickets)
 
-       # Takes node from alldownq, gets tickets.  
+       # Takes node from qin_toCheck, gets tickets.  
        # Thread that actually gets the tickets.
        def getTickets(self):
+               self.count = 0
                while 1:
-                       host = self.bucket.get(block = True)
-                       if host == "None": break
-                       #if self.tickets.has_key(host) == False:
-                       logger.debug("Popping from q - %s" %host)
-                       tmp = rt_tickets(host)
-                       if tmp:
-                               logger.debug("Found tickets for %s" %host)
-                               self.tickets[host] = tmp 
+                       diag_node = self.qin_toCheck.get(block = True)
+                       if diag_node == "None": 
+                               print "RT processed %d nodes with noticket" % self.count
+                               logger.debug("RT filtered %d noticket nodes" % self.count)
+                               self.qout_sickNoTicket.put("None")
+                               break
+                       else:
+                               host = diag_node['nodename']
+                               (b_host_inticket, r_ticket) = is_host_in_rt_tickets(host, \
+                                                                                                       self.l_ticket_blacklist, \
+                                                                                                       self.dbTickets)
+                               if b_host_inticket:
+                                       logger.debug("RT: found tickets for %s" %host)
+                                       diag_node['stage'] = 'stage_rt_working'
+                                       diag_node['ticket_id'] = r_ticket['ticket_id']
+                                       self.tickets[host] = r_ticket
+                               else:
+                                       #logger.debug("RT: no tix for %s" %host)
+                                       #print "no tix for %s" % host
+                                       if r_ticket is not None:
+                                               print "Ignoring ticket %s" % r_ticket['ticket_id']
+                                       self.count = self.count + 1
+
+                               # process diag_node for either case
+                               self.qout_sickNoTicket.put(diag_node) 
 
        # Removes hosts that are no longer down.
        def remTickets(self):
@@ -176,14 +254,14 @@ class RT(Thread):
                for host in prevdown:
                        if host not in currdown:
                                del self.tickets[host]
-                               logger.info("%s no longer down" % host)
+                               logger.info("RT: %s no longer down." % host)
 
        # Update Tickets
        def updateTickets(self):
                logger.info("Refreshing DB.")
                for host in self.tickets.keys():
                        # Put back in Q to refresh
-                       self.bucket.put(host)
+                       self.qin_toCheck.put(host)
 
        def cleanTickets(self):
                while 1:
@@ -210,7 +288,7 @@ def main():
        b.start()
        c.start()
        d.start()
-       tmp = ('planetlab-2.vuse.vanderbilt.edu', 'planetlab-11.cs.princeton.edu', 'planet03.csc.ncsu.edu', 'planetlab1.pop-rj.rnp.br', 'planet1.halifax.canet4.nodes.planet-lab.org', 'planet1.cavite.nodes.planet-lab.org', 'ds-pl3.technion.ac.il', 'planetlab2.cs.purdue.edu', 'planetlab3.millennium.berkeley.edu', 'planetlab1.unl.edu', 'planetlab1.cs.colorado.edu', 'planetlab02.cs.washington.edu', 'orbpl2.rutgers.edu', 'planetlab2.informatik.uni-erlangen.de', 'pl2.ernet.in', 'neu2.6planetlab.edu.cn', 'planetlab-2.cs.uni-paderborn.de', 'planetlab1.elet.polimi.it', 'planetlab2.iiitb.ac.in', 'server1.planetlab.iit-tech.net', 'planetlab2.iitb.ac.in', 'planetlab1.ece.ucdavis.edu', 'planetlab02.dis.unina.it', 'planetlab-1.dis.uniroma1.it', 'planetlab1.iitb.ac.in', 'pku1.6planetlab.edu.cn', 'planetlab1.warsaw.rd.tp.pl', 'planetlab2.cs.unc.edu', 'csu2.6planetlab.edu.cn', 'pl1.ernet.in', 'planetlab2.georgetown.edu', 'planetlab1.cs.uchicago.edu') 
+       tmp = ('planetlab-1.cs.ucy.ac.cy','planetlab-2.vuse.vanderbilt.edu', 'planetlab-11.cs.princeton.edu', 'planet03.csc.ncsu.edu', 'planetlab1.pop-rj.rnp.br', 'planet1.halifax.canet4.nodes.planet-lab.org', 'planet1.cavite.nodes.planet-lab.org', 'ds-pl3.technion.ac.il', 'planetlab2.cs.purdue.edu', 'planetlab3.millennium.berkeley.edu', 'planetlab1.unl.edu', 'planetlab1.cs.colorado.edu', 'planetlab02.cs.washington.edu', 'orbpl2.rutgers.edu', 'planetlab2.informatik.uni-erlangen.de', 'pl2.ernet.in', 'neu2.6planetlab.edu.cn', 'planetlab-2.cs.uni-paderborn.de', 'planetlab1.elet.polimi.it', 'planetlab2.iiitb.ac.in', 'server1.planetlab.iit-tech.net', 'planetlab2.iitb.ac.in', 'planetlab1.ece.ucdavis.edu', 'planetlab02.dis.unina.it', 'planetlab-1.dis.uniroma1.it', 'planetlab1.iitb.ac.in', 'pku1.6planetlab.edu.cn', 'planetlab1.warsaw.rd.tp.pl', 'planetlab2.cs.unc.edu', 'csu2.6planetlab.edu.cn', 'pl1.ernet.in', 'planetlab2.georgetown.edu', 'planetlab1.cs.uchicago.edu') 
        for host in tmp:
                bucket.put(host)
        #et = Thread(target=e.pushHosts)