X-Git-Url: http://git.onelab.eu/?a=blobdiff_plain;f=policy.py;h=0bdf4bffb7fba55b76fe1cd1911db7fb6db38e26;hb=dc70043f70f08f0d6540c77e53d30c0ed6e58ad5;hp=2199d4614b39c67e3a2da8258afca898b0797cd3;hpb=689ae55c995a8a0a4b555e93c32acf0f6cb75315;p=monitor.git diff --git a/policy.py b/policy.py index 2199d46..0bdf4bf 100644 --- a/policy.py +++ b/policy.py @@ -3,7 +3,7 @@ # # Faiyaz Ahmed # -# $Id: policy.py,v 1.16 2007/08/08 13:30:42 soltesz Exp $ +# $Id: policy.py,v 1.17 2007/08/29 17:26:50 soltesz Exp $ # # Policy Engine. @@ -17,10 +17,11 @@ import pickle import Queue import plc import sys +import os import reboot import soltesz import string -from printbadbysite import cmpCategoryVal +from www.printbadnodes import cmpCategoryVal from config import config print "policy" config = config() @@ -40,6 +41,8 @@ SLICEMAIL="%s@slices.planet-lab.org" PLCEMAIL="support@planet-lab.org" #Thresholds (DAYS) +SPERMIN = 60 +SPERHOUR = 60*60 SPERDAY = 86400 PITHRESH = 7 * SPERDAY SLICETHRESH = 7 * SPERDAY @@ -128,7 +131,14 @@ class Merge(Thread): fb_record = {} fb_record['nodename'] = nodename - fb_record['category'] = values['category'] + try: + fb_record['category'] = values['category'] + except: + print values + print nodename + print self.findbad['nodes'][nodename] + count -= 1 + continue fb_record['state'] = values['state'] fb_record['comonstats'] = values['comonstats'] fb_record['plcnode'] = values['plcnode'] @@ -195,27 +205,37 @@ class Merge(Thread): self.mergedb[loginbase][nodename] = {} self.mergedb[loginbase][nodename].update(x) self.mergedb[loginbase][nodename]['ticket_id'] = "" - self.mergedb[loginbase][nodename]['prev_category'] = None + self.mergedb[loginbase][nodename]['prev_category'] = "NORECORD" else: + if len(self.act_all[nodename]) == 0: + print "len(act_all[%s]) == 0, skipping %s %s" % (nodename, loginbase, nodename) + continue + y = self.act_all[nodename][0] - # skip if end-stage - if 'stage' in y and "monitor-end-record" in y['stage']: - continue + ## skip if end-stage + #if 'stage' in y and "monitor-end-record" in y['stage']: + # # 1) ok, b/c it's a new problem. set ticket_id to null + ## self.mergedb[loginbase][nodename] = {} + # self.mergedb[loginbase][nodename].update(x) + # self.mergedb[loginbase][nodename]['ticket_id'] = "" + # self.mergedb[loginbase][nodename]['prev_category'] = None + # continue + + ## for legacy actions + #if 'bucket' in y and y['bucket'][0] == 'dbg': + # # Only bootcd debugs made it to the act_all db. + # y['prev_category'] = "OLDBOOTCD" + #elif 'bucket' in y and y['bucket'][0] == 'down': + # y['prev_category'] = "ERROR" + #elif 'bucket' not in y: + # # for all other actions, just carry over the + # # previous category + # y['prev_category'] = y['category'] + #else: + # print "UNKNOWN state for record: %s" % y + # sys.exit(1) - # for legacy actions - if 'bucket' in y and y['bucket'][0] == 'dbg': - # Only bootcd debugs made it to the act_all db. - y['prev_category'] = "OLDBOOTCD" - elif 'bucket' in y and y['bucket'][0] == 'down': - y['prev_category'] = "ERROR" - elif 'bucket' not in y: - # for all other actions, just carry over the - # previous category - y['prev_category'] = y['category'] - else: - print "UNKNOWN state for record: %s" % y - sys.exit(1) # determine through translation, if the buckets match #if 'category' in y and x['category'] == y['category']: # b_match = True @@ -229,7 +249,6 @@ class Merge(Thread): #if b_match: # # 2b) ok, b/c they agree that there's still a problem.. # # 2b) Comon & Monitor still agree; RT ticket? - # y['prev_category'] = y['category'] #else: # # 2a) mismatch, need a policy for how to resolve # # resolution will be handled in __diagnoseNode() @@ -239,7 +258,7 @@ class Merge(Thread): # print "FINDBAD and MONITOR have a mismatch: %s vs %s" % \ # (x['category'], y['bucket']) - + y['prev_category'] = y['category'] self.mergedb[loginbase][nodename] = {} self.mergedb[loginbase][nodename].update(y) self.mergedb[loginbase][nodename]['comonstats'] = x['comonstats'] @@ -407,7 +426,7 @@ class Diagnose(Thread): d_diag_nodes are diagnose_in entries. """ d_diag_site = {loginbase : { 'config' : - {'squeeze': False, + {'squeeze': False, 'email': False }, 'nodes': {} @@ -421,6 +440,16 @@ class Diagnose(Thread): if diag_record != None: d_diag_site[loginbase]['nodes'][nodename] = diag_record + + # NOTE: improvement means, we need to act/squeeze and email. + #print "DIAG_RECORD", diag_record + if 'monitor-end-record' in diag_record['stage'] or \ + 'nmreset' in diag_record['stage']: + # print "resetting loginbase!" + d_diag_site[loginbase]['config']['squeeze'] = True + d_diag_site[loginbase]['config']['email'] = True + #else: + # print "NO IMPROVEMENT!!!!" else: pass # there is nothing to do for this node. @@ -462,6 +491,17 @@ class Diagnose(Thread): diag_record['message'] = emailTxt.mailtxt.newdown diag_record['args'] = {'nodename': nodename} diag_record['info'] = (nodename, s_daysdown, "") + + if 'reboot_node_failed' in node_record: + # there was a previous attempt to use the PCU. + if node_record['reboot_node_failed'] == False: + # then the last attempt apparently, succeeded. + # But, the category is still 'ERROR'. Therefore, the + # PCU-to-Node mapping is broken. + #print "Setting message for ERROR node to PCU2NodeMapping: %s" % nodename + diag_record['message'] = emailTxt.mailtxt.pcutonodemapping + diag_record['email_pcu'] = True + if diag_record['ticket_id'] == "": diag_record['log'] = "DOWN: %20s : %-40s == %20s %s" % \ (loginbase, nodename, diag_record['info'][1:], diag_record['found_rt_ticket']) @@ -497,6 +537,13 @@ class Diagnose(Thread): elif "BOOT" in state: # no action needed. # TODO: remove penalties, if any are applied. + now = time.time() + last_contact = node_record['plcnode']['last_contact'] + if last_contact == None: + time_diff = 0 + else: + time_diff = now - last_contact; + if 'improvement' in node_record['stage']: # then we need to pass this on to 'action' diag_record = {} @@ -514,6 +561,27 @@ class Diagnose(Thread): (loginbase, nodename, diag_record['stage'], state, category, diag_record['ticket_id']) return diag_record + elif time_diff >= 6*SPERHOUR: + # heartbeat is older than 30 min. + # then reset NM. + #print "Possible NM problem!! %s - %s = %s" % (now, last_contact, time_diff) + diag_record = {} + diag_record.update(node_record) + diag_record['message'] = emailTxt.mailtxt.NMReset + diag_record['args'] = {'nodename': nodename} + diag_record['stage'] = "nmreset" + diag_record['info'] = (nodename, + node_record['prev_category'], + node_record['category']) + if diag_record['ticket_id'] == "": + diag_record['log'] = "NM : %20s : %-40s == %20s %20s %s %s" % \ + (loginbase, nodename, diag_record['stage'], + state, category, diag_record['found_rt_ticket']) + else: + diag_record['log'] = "NM : %20s : %-40s == %20s" % \ + (loginbase, nodename, diag_record['stage']) + + return diag_record else: return None else: @@ -536,36 +604,44 @@ class Diagnose(Thread): def __diagnoseNode(self, loginbase, node_record): # TODO: change the format of the hostname in this # record to something more natural. - nodename = node_record['nodename'] - category = node_record['category'] - prev_category = node_record['prev_category'] - state = node_record['state'] - - val = cmpCategoryVal(category, prev_category) - if val == -1: - # current category is worse than previous, carry on - pass - elif val == 1: - # current category is better than previous - # TODO: too generous for now, but will be handled correctly - # TODO: if stage is currently ticket_waitforever, - if 'ticket_id' not in node_record: - print "ignoring: ", node_record['nodename'] - return None - else: - if node_record['ticket_id'] == "" or \ - node_record['ticket_id'] == None: - print "closing: ", node_record['nodename'] + nodename = node_record['nodename'] + category = node_record['category'] + prev_category = node_record['prev_category'] + state = node_record['state'] + #if 'prev_category' in node_record: + # prev_category = node_record['prev_category'] + #else: + # prev_category = "ERROR" + if node_record['prev_category'] != "NORECORD": + + val = cmpCategoryVal(category, prev_category) + print "%s went from %s -> %s" % (nodename, prev_category, category) + if val == 1: + # improved + if node_record['ticket_id'] == "" or node_record['ticket_id'] == None: + print "closing record with no ticket: ", node_record['nodename'] node_record['action'] = ['close_rt'] node_record['message'] = None node_record['stage'] = 'monitor-end-record' return node_record - #return None else: node_record['stage'] = 'improvement' - else: - #values are equal, carry on. - pass + + #if 'monitor-end-record' in node_record['stage']: + # # just ignore it if it's already ended. + # # otherwise, the status should be worse, and we won't get + # # here. + # print "monitor-end-record: ignoring ", node_record['nodename'] + # return None +# +# #return None + elif val == -1: + # current category is worse than previous, carry on + pass + else: + #values are equal, carry on. + #print "why are we here?" + pass #### COMPARE category and prev_category # if not_equal @@ -574,6 +650,7 @@ class Diagnose(Thread): # then check category for stats. diag_record = self.diagRecordByCategory(node_record) if diag_record == None: + #print "diag_record == None" return None #### found_RT_ticket @@ -588,21 +665,34 @@ class Diagnose(Thread): # TODO: generalize delays at PLC, and prevent enforcement when there # have been no emails. # NOTE: 7*SPERDAY exists to offset the 'bad week' - delta = current_time - diag_record['time'] - 7*SPERDAY + #delta = current_time - diag_record['time'] - 7*SPERDAY + delta = current_time - diag_record['time'] message = diag_record['message'] act_record = {} act_record.update(diag_record) #### DIAGNOSE STAGES - #print "%s has stage %s" % (nodename, diag_record['stage']) if 'findbad' in diag_record['stage']: # The node is bad, and there's no previous record of it. - act_record['email'] = TECH # addative emails + act_record['email'] = TECH act_record['action'] = ['noop'] act_record['message'] = message[0] act_record['stage'] = 'stage_actinoneweek' + elif 'nmreset' in diag_record['stage']: + act_record['email'] = ADMIN + act_record['action'] = ['reset_nodemanager'] + act_record['message'] = message[0] + act_record['stage'] = 'nmreset' + return None + + elif 'reboot_node' in diag_record['stage']: + act_record['email'] = TECH + act_record['action'] = ['noop'] + act_record['message'] = message[0] + act_record['stage'] = 'stage_actinoneweek' + elif 'improvement' in diag_record['stage']: # - backoff previous squeeze actions (slice suspend, nocreate) # TODO: add a backoff_squeeze section... Needs to runthrough @@ -616,6 +706,7 @@ class Diagnose(Thread): act_record['stage'] = 'stage_actintwoweeks' act_record['message'] = message[1] act_record['action'] = ['nocreate' ] + act_record['time'] = current_time # reset clock for waitforever elif delta >= 3* SPERDAY and not 'second-mail-at-oneweek' in act_record: act_record['email'] = TECH act_record['message'] = message[0] @@ -624,16 +715,17 @@ class Diagnose(Thread): else: act_record['message'] = None act_record['action'] = ['waitforoneweekaction' ] + print "ignoring this record for: %s" % act_record['nodename'] return None # don't send if there's no action elif 'actintwoweeks' in diag_record['stage']: - if delta >= 14 * SPERDAY: + if delta >= 7 * SPERDAY: act_record['email'] = TECH | PI | USER act_record['stage'] = 'stage_waitforever' act_record['message'] = message[2] act_record['action'] = ['suspendslices'] act_record['time'] = current_time # reset clock for waitforever - elif delta >= 10* SPERDAY and not 'second-mail-at-twoweeks' in act_record: + elif delta >= 3* SPERDAY and not 'second-mail-at-twoweeks' in act_record: act_record['email'] = TECH | PI act_record['message'] = message[1] act_record['action'] = ['sendmailagain-waitfortwoweeksaction' ] @@ -681,11 +773,12 @@ class Diagnose(Thread): # 1. stage is unknown, or # 2. delta is not big enough to bump it to the next stage. # TODO: figure out which. for now assume 2. - print "UNKNOWN!!? %s" % nodename + print "UNKNOWN stage for %s; nothing done" % nodename act_record['action'] = ['unknown'] act_record['message'] = message[0] - print "Exiting..." - sys.exit(1) + #print "Exiting..." + return None + #sys.exit(1) print "%s" % act_record['log'], print "%15s" % act_record['action'] @@ -783,6 +876,14 @@ def close_rt_backoff(args): plc.enableSliceCreation(args['hostname']) return +def reboot_node(args): + host = args['hostname'] + return reboot.reboot_new(host, True, config.debug) + +def reset_nodemanager(args): + os.system("ssh root@%s /sbin/service nm restart" % nodename) + return + class Action(Thread): def __init__(self, l_action): self.l_action = l_action @@ -802,6 +903,9 @@ class Action(Thread): self.actions['close_rt'] = lambda args: close_rt_backoff(args) self.actions['rins'] = lambda args: plc.nodeBootState(args['hostname'], "rins") self.actions['noop'] = lambda args: args + self.actions['reboot_node'] = lambda args: reboot_node(args) + self.actions['reset_nodemanager'] = lambda args: args # reset_nodemanager(args) + self.actions['ticket_waitforever'] = lambda args: args self.actions['waitforever'] = lambda args: args self.actions['unknown'] = lambda args: args @@ -922,14 +1026,6 @@ class Action(Thread): mailer.email(subj, body, contacts) ticket_id = args['ticket_id'] else: - #if 'ticket_id' in args and 'ticket_id' != "": - # # Reformat Subject to include Ticket_ID for RT - # subj = "Re: [PL #%s] %s" % (args['ticket_id'], subject) - # # RT remembers old contacts, so only add new users - # mailer.email(subj, body, ['monitor@planet-lab.org'] + contacts) - # ticket_id = args['ticket_id'] - #else: - # ticket_id = mailer.emailViaRT(subject, body, contacts) ticket_id = mailer.emailViaRT(subject, body, contacts, args['ticket_id']) except Exception, err: print "exception on message:" @@ -957,6 +1053,14 @@ class Action(Thread): for act_record in act_recordlist: email_args['hostname_list'] += act_record['msg_format'] email_args['hostname'] = act_record['nodename'] + if 'plcnode' in act_record and \ + 'pcu_ids' in act_record['plcnode'] and \ + len(act_record['plcnode']['pcu_ids']) > 0: + print "setting 'pcu_id' for email_args %s"%email_args['hostname'] + email_args['pcu_id'] = act_record['plcnode']['pcu_ids'][0] + else: + email_args['pcu_id'] = "-1" + if 'ticket_id' in act_record: email_args['ticket_id'] = act_record['ticket_id'] @@ -985,7 +1089,8 @@ class Action(Thread): diag_record = site_record['nodes'][nodename] act_record = self.__actOnNode(diag_record) #print "nodename: %s %s" % (nodename, act_record) - act_recordlist += [act_record] + if act_record is not None: + act_recordlist += [act_record] unique_issues = self.get_unique_issues(act_recordlist) @@ -993,6 +1098,21 @@ class Action(Thread): print "\tworking on issue: %s" % issue issue_record_list = unique_issues[issue] email_args = self.get_email_args(issue_record_list) + + # for each record. + for act_record in issue_record_list: + # if there's a pcu record and email config is set + if 'email_pcu' in act_record: + if act_record['email_pcu'] and \ + site_record['config']['email']: + + email_args['hostname'] = act_record['nodename'] + ticket_id = self.__emailSite(loginbase, + act_record['email'], + emailTxt.mailtxt.pcudown[0], + email_args) + email_args['ticket_id'] = ticket_id + act_record = issue_record_list[0] # send message before squeezing @@ -1006,16 +1126,16 @@ class Action(Thread): for act_record in issue_record_list: nodename = act_record['nodename'] # update node record with RT ticket_id - self.act_all[nodename][0]['ticket_id'] = "%s" % ticket_id + if nodename in self.act_all: + self.act_all[nodename][0]['ticket_id'] = "%s" % ticket_id if config.mail: i_nodes_emailed += 1 print "\t\tconfig.squeeze: %s and %s" % (config.squeeze, site_record['config']['squeeze']) if config.squeeze and site_record['config']['squeeze']: for act_key in act_record['action']: - #act_key = act_record['action'] self.actions[act_key](email_args) - i_nodes_actedon += 1 + i_nodes_actedon += 1 if config.policysavedb: print "Saving Databases... act_all, diagnose_out" @@ -1040,11 +1160,60 @@ class Action(Thread): act_record.update(diag_record) act_record['nodename'] = nodename act_record['msg_format'] = self._format_diaginfo(diag_record) + print "act_record['stage'] == %s " % act_record['stage'] + + # avoid end records, and nmreset records + # reboot_node_failed, is set below, so don't reboot repeatedly. + + if 'monitor-end-record' not in act_record['stage'] and \ + 'nmreset' not in act_record['stage'] and \ + 'reboot_node_failed' not in act_record: + + if "DOWN" in act_record['log'] and \ + 'pcu_ids' in act_record['plcnode'] and \ + len(act_record['plcnode']['pcu_ids']) > 0: + + print "%s" % act_record['log'], + print "%15s" % (['reboot_node'],) + # Set node to re-install + plc.nodeBootState(act_record['nodename'], "rins") + try: + ret = reboot_node({'hostname': act_record['nodename']}) + except Exception, exc: + print "exception on reboot_node:" + import traceback + print traceback.print_exc() + ret = False + + if ret: # and ( 'reboot_node_failed' not in act_record or act_record['reboot_node_failed'] == False): + # Reboot Succeeded + print "reboot succeeded for %s" % act_record['nodename'] + act_record2 = {} + act_record2.update(act_record) + act_record2['action'] = ['reboot_node'] + act_record2['stage'] = "reboot_node" + act_record2['reboot_node_failed'] = False + act_record2['email_pcu'] = False + + if nodename not in self.act_all: + self.act_all[nodename] = [] + print "inserting 'reboot_node' record into act_all" + self.act_all[nodename].insert(0,act_record2) + + # return None to avoid further action + print "Taking no further action" + return None + else: + print "reboot failed for %s" % act_record['nodename'] + # set email_pcu to also send pcu notice for this record. + act_record['reboot_node_failed'] = True + act_record['email_pcu'] = True - print "%s" % act_record['log'], - print "%15s" % act_record['action'] + print "%s" % act_record['log'], + print "%15s" % act_record['action'] - if act_record['stage'] is not 'monitor-end-record': + if act_record['stage'] is not 'monitor-end-record' and \ + act_record['stage'] is not 'nmreset': if nodename not in self.act_all: self.act_all[nodename] = []