HADOOP-3184. Modified HOD to handle master failures on bad nodes by trying to bring them up on another node in the ring. Contributed by Hemanth Yamijala.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/core/trunk@663889 13f79535-47bb-0310-9956-ffa450edef68

src/contrib/hod/CHANGES.txt

@@ -9,6 +9,10 @@ Trunk (unreleased changes)
     exist and to auto-deallocate a cluster while reallocating it, if it is
     already dead. (Hemanth Yamijala via mukund)
 
+    HADOOP-3184. Modified HOD to handle master failures on bad nodes by trying 
+    to bring them up on another node in the ring. (Hemanth Yamijala via ddas)
+
+
   NEW FEATURES
 
   IMPROVEMENTS

src/contrib/hod/bin/hod

@@ -225,7 +225,12 @@ defList = { 'hod' : (
              False, 120, False, True),
 
              ('idleness-limit', 'pos_int', 'Limit after which to deallocate the cluster',
-             False, 3600, False, True)),
+             False, 3600, False, True),
+
+             ('max-master-failures', 'pos_int', 
+              'Defines how many times a master can fail before' \
+              ' failing cluster allocation', False, 5, True, True)),
+
 
             'gridservice-mapred' : (
              ('external', 'bool', "Connect to an already running MapRed?",

src/contrib/hod/bin/ringmaster

@@ -113,7 +113,11 @@ defList = { 'ringmaster' : (
              False, 120, False, True),
 
              ('idleness-limit', 'pos_int', 'Limit after which to deallocate the cluster',
-             False, 3600, False, True)),
+             False, 3600, False, True),
+
+             ('max-master-failures', 'pos_int', 
+              'Defines how many times a master can fail before' \
+              ' failing cluster allocation', False, 5, True, True)),
 
             'resource_manager' : (
              ('id', 'string', 'Batch scheduler ID: torque|condor.',
@@ -330,7 +334,8 @@ if __name__ == '__main__':
                                      'ringmaster', 'svcrgy-addr'))
       serviceClient = hodXRClient(serviceAddr)
       if serviceClient is not None:
-        serviceClient.setRMError([str(e),get_exception_string()])
+        serviceClient.setRMError([local_fqdn(), str(e), \
+                                    get_exception_string()])
         log.info("Reported errors to service registry at %s" % serviceAddr)
     except Exception, e:
       log.error("Failed to report errors to service registry.")

src/contrib/hod/hodlib/GridServices/service.py

@@ -66,6 +66,7 @@ class MasterSlave(Service):
     self.masterAddress = 'none'
     self.requiredNode = requiredNode
     self.failedMsg = None
+    self.masterFailureCount = 0
 
   def getRequiredNode(self):
     return self.requiredNode
@@ -136,6 +137,9 @@ class MasterSlave(Service):
     """ set the master initialized to
     true. """
     self.masterInitialized = True
+    # Reset failure related variables, as master is initialized successfully.
+    self.masterFailureCount = 0
+    self.failedMsg = None
 
   def getMasterAddress(self):
     """ it needs to change to reflect 
@@ -152,11 +156,19 @@ class MasterSlave(Service):
     return self.serviceDesc.isExternal()
 
   def setMasterFailed(self, err):
+    """Sets variables related to Master failure"""
+    self.masterFailureCount += 1
     self.failedMsg = err
+    # When command is sent to HodRings, this would have been set to True.
+    # Reset it to reflect the correct status.
+    self.launchedMaster = False
 
   def getMasterFailed(self):
     return self.failedMsg
-  
+ 
+  def getMasterFailureCount(self):
+    return self.masterFailureCount
+ 
 class NodeRequest:
   """ A class to define 
   a node request. """

src/contrib/hod/hodlib/Hod/hadoop.py

@@ -604,9 +604,11 @@ class hadoopCluster:
     if status == 5 or status == 6:
       ringMasterErrors = self.__svcrgyClient.getRMError()
       if ringMasterErrors:
-        self.__log.critical("Cluster could not be allocated because of the following errors on the ringmaster host.\n%s" % \
-                               (ringMasterErrors[0]))
-        self.__log.debug("Stack trace on ringmaster: %s" % ringMasterErrors[1])
+        self.__log.critical("Cluster could not be allocated because" \
+                            " of the following errors on the "\
+                            "ringmaster host %s.\n%s" % \
+                               (ringMasterErrors[0], ringMasterErrors[1]))
+        self.__log.debug("Stack trace on ringmaster: %s" % ringMasterErrors[2])
     return status
 
   def __isRingMasterAlive(self, rmAddr):

src/contrib/hod/hodlib/Hod/hod.py

@@ -227,6 +227,31 @@ class hodRunner:
          
     return opList 
   
+  def __adjustMasterFailureCountConfig(self, nodeCount):
+    # This method adjusts the ringmaster.max-master-failures variable
+    # to a value that is bounded by the a function of the number of
+    # nodes.
+
+    maxFailures = self.__cfg['ringmaster']['max-master-failures']
+    # Count number of masters required - depends on which services
+    # are external
+    masters = 0
+    if not self.__cfg['gridservice-hdfs']['external']:
+      masters += 1
+    if not self.__cfg['gridservice-mapred']['external']:
+      masters += 1
+
+    # So, if there are n nodes and m masters, we look atleast for
+    # all masters to come up. Therefore, atleast m nodes should be
+    # good, which means a maximum of n-m master nodes can fail.
+    maxFailedNodes = nodeCount - masters
+
+    # The configured max number of failures is now bounded by this
+    # number.
+    self.__cfg['ringmaster']['max-master-failures'] = \
+                              min(maxFailures, maxFailedNodes)
+
+    
   def _op_allocate(self, args):
     operation = "allocate"
     argLength = len(args)
@@ -312,6 +337,9 @@ class hodRunner:
               self.__cleanup()
               raise HodInterruptException()
             self.__log.debug("Service Registry started.")
+
+            self.__adjustMasterFailureCountConfig(nodes)
+            
             try:
               allocateStatus = self.__cluster.allocate(clusterDir, min, max)    
             except HodInterruptException, h:

src/contrib/hod/hodlib/RingMaster/ringMaster.py

@@ -370,7 +370,12 @@ class _LogMasterSources:
         for v in self.serviceDict.itervalues():
           if (not v.isExternal()):
             if v.isLaunchable(self.serviceDict):
-              if not v.isMasterLaunched():
+              # If a master is still not launched, or the number of 
+              # retries for launching master is not reached, 
+              # launch master
+              if not v.isMasterLaunched() and \
+                  (v.getMasterFailureCount() <= \
+                      self.cfg['ringmaster']['max-master-failures']):
                 cmdList = v.getMasterCommands(self.serviceDict)
                 v.setlaunchedMaster()
                 v.setMasterAddress(addr)
@@ -441,7 +446,8 @@ class _LogMasterSources:
   def setHodRingErrors(self, addr, errors):
     """This method is called by the hodrings to update errors 
       it encountered while starting up"""
-    self.log.critical("Hodring at %s failed with following errors:\n%s" % (addr, errors))
+    self.log.critical("Hodring at %s failed with following errors:\n%s" \
+                        % (addr, errors))
     lock = self.masterParamLock
     lock.acquire()
     try:
@@ -452,7 +458,8 @@ class _LogMasterSources:
             idx = addr.rfind('_')
             if idx is not -1:
               addr = addr[:idx]
-            v.setMasterFailed("Hodring at %s failed with following errors:\n%s" % (addr, errors))
+            v.setMasterFailed("Hodring at %s failed with following" \
+                                " errors:\n%s" % (addr, errors))
     except:
       self.log.debug(get_exception_string())
       pass
@@ -478,8 +485,16 @@ class _LogMasterSources:
       pass
     else:
       self.log.debug("getServiceAddr service: %s" % service)
+      # Check if we should give up ! If the limit on max failures is hit, 
+      # give up.
       err = service.getMasterFailed()
-      if err is not None:
+      if (err is not None) and \
+            (service.getMasterFailureCount() > \
+                      self.cfg['ringmaster']['max-master-failures']):
+        self.log.critical("Detected errors (%s) beyond allowed number"\
+                            " of failures (%s). Flagging error to client" \
+                            % (service.getMasterFailureCount(), \
+                              self.cfg['ringmaster']['max-master-failures']))
         addr = "Error: " + err
       elif (service.isMasterInitialized()):
         addr = service.getMasterAddrs()[0]