path: root/third_party/hadoop-0.20.0/contrib/hod/hodlib/RingMaster/ringMaster.py

#Licensed to the Apache Software Foundation (ASF) under one
#or more contributor license agreements.  See the NOTICE file
#distributed with this work for additional information
#regarding copyright ownership.  The ASF licenses this file
#to you under the Apache License, Version 2.0 (the
#"License"); you may not use this file except in compliance
#with the License.  You may obtain a copy of the License at

#     http://www.apache.org/licenses/LICENSE-2.0

#Unless required by applicable law or agreed to in writing, software
#distributed under the License is distributed on an "AS IS" BASIS,
#WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#See the License for the specific language governing permissions and
#limitations under the License.
#!/usr/bin/env python
"""manages services and nodepool"""
# -*- python -*-

import os, sys, random, time, sets, shutil, threading
import urllib, urlparse, re, getpass, pprint, signal, shutil

from pprint import pformat
from HTMLParser import HTMLParser

binfile = sys.path[0]
libdir = os.path.dirname(binfile)
sys.path.append(libdir)

import hodlib.Common.logger
from hodlib.RingMaster.idleJobTracker import JobTrackerMonitor, HadoopJobStatus

from hodlib.Common.threads import func 

from hodlib.Hod.nodePool import *
from hodlib.Common.util import *
from hodlib.Common.nodepoolutil import NodePoolUtil
from hodlib.Common.socketServers import hodXMLRPCServer
from hodlib.Common.socketServers import threadedHTTPServer
from hodlib.NodePools import *
from hodlib.NodePools.torque import *
from hodlib.GridServices import *
from hodlib.Common.descGenerator import *
from hodlib.Common.xmlrpc import hodXRClient
from hodlib.Common.miniHTMLParser import miniHTMLParser
from hodlib.Common.threads import simpleCommand

class ringMasterServer:
  """The RPC server that exposes all the master config
  changes. Also, one of these RPC servers runs as a proxy
  and all the hodring instances register with this proxy"""
  instance = None
  xmlrpc = None
  
  def __init__(self, cfg, log, logMasterSources, retry=5):
    try:
      from hodlib.Common.socketServers import twistedXMLRPCServer
      ringMasterServer.xmlrpc = twistedXMLRPCServer("", 
        cfg['ringmaster']['xrs-port-range'])
    except ImportError:
      log.info("Twisted interface not found. Using hodXMLRPCServer.")
      ringMasterServer.xmlrpc = hodXMLRPCServer("", 
        cfg['ringmaster']['xrs-port-range'])

    ringMasterServer.xmlrpc.register_instance(logMasterSources)
    self.logMasterSources = logMasterSources
    ringMasterServer.xmlrpc.serve_forever()
        
    while not ringMasterServer.xmlrpc.is_alive():
      time.sleep(.5)
          
    log.debug('Ringmaster RPC Server at %d' % 
                 ringMasterServer.xmlrpc.server_address[1])
    
  def startService(ss, cfg, np, log, rm):
    logMasterSources = _LogMasterSources(ss, cfg, np, log, rm)
    ringMasterServer.instance = ringMasterServer(cfg, log, logMasterSources)

  def stopService():
    ringMasterServer.xmlrpc.stop()
  
  def getPort():
    return ringMasterServer.instance.port

  def getAddress():
    return 'http://%s:%d/' % (socket.gethostname(), 
                              ringMasterServer.xmlrpc.server_address[1])
  
  startService = staticmethod(startService)
  stopService = staticmethod(stopService)
  getPort = staticmethod(getPort)
  getAddress = staticmethod(getAddress)
  
class _LogMasterSources:
  """All the methods that are run by the RPC server are
  added into this class """
  
  def __init__(self, serviceDict, cfg, np, log, rm):
    self.serviceDict = serviceDict
    self.tarSource = []
    self.tarSourceLock = threading.Lock()
    self.dict = {}
    self.count = {}
    self.logsourceList = []
    self.logsourceListLock = threading.Lock()
    self.masterParam = []
    self.masterParamLock = threading.Lock()
    self.verify = 'none'
    self.cmdLock = threading.Lock()
    self.cfg = cfg
    self.log = log
    self.np = np
    self.rm = rm 
    self.hdfsHost = None
    self.mapredHost = None
    self.maxconnect = self.cfg['ringmaster']['max-connect']
    self.log.debug("Using max-connect value %s"%self.maxconnect)

   
  def registerTarSource(self, hostname, url, addr=None):
    self.log.debug("registering: " + url)
    lock = self.tarSourceLock
    lock.acquire()
    self.dict[url] = url
    self.count[url] = 0
    # addr is None when ringMaster himself invokes this method
    if addr:
      c = self.count[addr]
      self.count[addr] = c - 1
    lock.release()
    if addr:
      str = "%s is done" % (addr)
      self.log.debug(str)
    return url

  def getTarList(self,hodring):   # this looks useful
    lock = self.tarSourceLock
    lock.acquire()
    leastkey = None
    leastval = -1
    for k, v in self.count.iteritems():
      if (leastval  == -1):
        leastval = v
        pass
      if (v <= leastval and v < self.maxconnect):
        leastkey = k
        leastval = v
    if (leastkey == None):
      url  = 'none'
    else:
      url = self.dict[leastkey]
      self.count[leastkey] = leastval + 1
      self.log.debug("%s %d" % (leastkey, self.count[leastkey]))
    lock.release()
    self.log.debug('sending url ' + url+" to "+hodring)  # this looks useful
    return url

  def tarDone(self, uri):
    str = "%s is done" % (uri)
    self.log.debug(str)
    lock = self.tarSourceLock
    lock.acquire()
    c = self.count[uri]
    self.count[uri] = c - 1
    lock.release()
    return uri

  def status(self):
    return True

# FIXME: this code is broken, it relies on a central service registry
#
#  def clusterStart(self, changedClusterParams=[]):
#    self.log.debug("clusterStart method invoked.")
#    self.dict = {}
#    self.count = {}
#    try:
#      if (len(changedClusterParams) > 0):
#        self.log.debug("Updating config.")
#        for param in changedClusterParams:
#          (key, sep1, val) = param.partition('=')
#          (i1, sep2, i2) = key.partition('.')
#          try:
#            prev = self.cfg[i1][i2]
#            self.rm.cfg[i1][i2] = val
#            self.cfg[i1][i2] = val
#            self.log.debug("\nModified [%s][%s]=%s to [%s][%s]=%s" % (i1, i2, prev, i1, i2, val))
#          except KeyError, e:
#            self.log.info("Skipping %s as no such config parameter found in ringmaster" % param)
#        self.log.debug("Regenerating Service Description.")
#        dGen = DescGenerator(self.rm.cfg)
#        self.rm.cfg['servicedesc'] = dGen.createServiceDescDict()
#        self.cfg['servicedesc'] = self.rm.cfg['servicedesc']
#  
#      self.rm.tar = None
#      if self.rm.cfg['ringmaster'].has_key('hadoop-tar-ball'):
#        self.rm.download = True
#        self.rm.tar = self.rm.cfg['ringmaster']['hadoop-tar-ball']
#        self.log.debug("self.rm.tar=%s" % self.rm.tar)
# 
#      self.rm.cd_to_tempdir()
#
#      self.rm.tarAddress = None 
#      hostname = socket.gethostname()
#      if (self.rm.download):
#        self.rm.basename = os.path.basename(self.rm.tar)
#        dest = os.path.join(os.getcwd(), self.rm.basename)
#        src =  self.rm.tar  
#        self.log.debug("cp %s -> %s" % (src, dest))
#        shutil.copy(src, dest) 
#        self.rm.tarAddress = "%s%s" % (self.rm.httpAddress, self.rm.basename)
#        self.registerTarSource(hostname, self.rm.tarAddress)
#        self.log.debug("Registered new tarAddress %s" % self.rm.tarAddress)
#      else:
#        self.log.debug("Download not set.")
#      
#      if (self.rm.tar != None):
#        self.cfg['hodring']['download-addr'] = self.rm.tarAddress
#        self.rm.cfg['hodring']['download-addr'] = self.rm.tarAddress
#
#      sdl = self.rm.cfg['servicedesc']
#      workDirs = self.rm.getWorkDirs(self.rm.cfg, True)
#      hdfsDesc = sdl['hdfs']
#      hdfs = None
#      if hdfsDesc.isExternal():
#        hdfs = HdfsExternal(hdfsDesc, workDirs)
#      else:
#        hdfs = Hdfs(hdfsDesc, workDirs, 0, False, True)
#    
#      self.rm.serviceDict[hdfs.getName()] = hdfs
#      mrDesc = sdl['mapred']
#      mr = None
#      if mrDesc.isExternal():
#        mr = MapReduceExternal(mrDesc, workDirs)
#      else:
#        mr = MapReduce(mrDesc, workDirs, 1)
#      self.rm.serviceDict[mr.getName()] = mr
#
#      ringList = self.rm.serviceClient.getServiceInfo(self.cfg['hodring']['userid'],
#        self.np.getServiceId(), 'hodring', 'hod') 
#    
#      slaveList = ringList
#      hdfsringXRAddress = None
#      # Start HDFS Master - Step 1
#      if not hdfsDesc.isExternal():
#        masterFound = False
#        for ring in ringList:
#          ringXRAddress = ring['xrs']
#          if ringXRAddress == None:
#            raise Exception("Could not get hodring XML-RPC server address.")
#          if  (ringXRAddress.find(self.hdfsHost) != -1):
#            ringClient = hodXRClient(ringXRAddress, None, None, 0, 0, 0, False, 0)
#            hdfsringXRAddress = ringXRAddress
#            self.log.debug("Invoking clusterStart on " + ringXRAddress + " (HDFS Master)")
#            ringClient.clusterStart()
#            masterFound = True 
#            slaveList.remove(ring)
#            break
#        if not masterFound:
#          raise Exception("HDFS Master host not found")
#        while hdfs.getInfoAddrs() == None:
#          self.log.debug("Waiting for HDFS Master (Name Node) to register dfs.info.port")
#          time.sleep(1)
#
#      # Start MAPRED Master - Step 2
#      if not mrDesc.isExternal():
#        masterFound = False
#        for ring in ringList:
#          ringXRAddress = ring['xrs']
#          if ringXRAddress == None:
#            raise Exception("Could not get hodring XML-RPC server address.")
#          if (not mrDesc.isExternal() and ringXRAddress.find(self.mapredHost) != -1):
#            ringClient = hodXRClient(ringXRAddress, None, None, 0, 0, 0, False, 0)
#            self.log.debug("Invoking clusterStart on " + ringXRAddress + " (MAPRED Master)")
#            ringClient.clusterStart()
#            masterFound = True 
#            slaveList.remove(ring)
#            break
#        if not masterFound:
#          raise Excpetion("MAPRED Master host not found")
#        while mr.getInfoAddrs() == None:
#          self.log.debug("Waiting for MAPRED Master (Job Tracker) to register \
# mapred.job.tracker.info.port")
#          time.sleep(1)
#
#      # Start Slaves - Step 3 
#      for ring in slaveList:
#          ringXRAddress = ring['xrs']
#          if ringXRAddress == None:
#            raise Exception("Could not get hodring XML-RPC server address.")
#          ringClient = hodXRClient(ringXRAddress, None, None, 0, 0, 0, False, 0)
#          self.log.debug("Invoking clusterStart on " + ringXRAddress + " (Slaves)")
#          ringThread = func(name='hodring_slaves_start', functionRef=ringClient.clusterStart())
#          ring['thread'] = ringThread
#          ringThread.start()
#
#      for ring in slaveList:
#        ringThread = ring['thread']
#        if ringThread == None:
#          raise Exception("Could not get hodring thread (Slave).")
#        ringThread.join()
#        self.log.debug("Completed clusterStart on " + ring['xrs'] + " (Slave)")
#
#      # Run Admin Commands on HDFS Master - Step 4
#      if not hdfsDesc.isExternal():
#        if hdfsringXRAddress == None:
#          raise Exception("HDFS Master host not found (to Run Admin Commands)")
#        ringClient = hodXRClient(hdfsringXRAddress, None, None, 0, 0, 0, False, 0)
#        self.log.debug("Invoking clusterStart(False) - Admin on "
#                       + hdfsringXRAddress + " (HDFS Master)")
#        ringClient.clusterStart(False)
#
#    except:
#      self.log.debug(get_exception_string())
#      return False
#
#    self.log.debug("Successfully started cluster.")
#    return True
#
#  def clusterStop(self):
#    self.log.debug("clusterStop method invoked.")
#    try:
#      hdfsAddr = self.getServiceAddr('hdfs')
#      if hdfsAddr.find(':') != -1:
#        h, p = hdfsAddr.split(':', 1)
#        self.hdfsHost = h
#        self.log.debug("hdfsHost: " + self.hdfsHost)
#      mapredAddr = self.getServiceAddr('mapred')
#      if mapredAddr.find(':') != -1:
#        h, p = mapredAddr.split(':', 1)
#        self.mapredHost = h
#        self.log.debug("mapredHost: " + self.mapredHost)
#      ringList = self.rm.serviceClient.getServiceInfo(self.cfg['hodring']['userid'],
#                                                      self.np.getServiceId(),
#                                                      'hodring', 'hod')
#      for ring in ringList:
#        ringXRAddress = ring['xrs']
#        if ringXRAddress == None:
#          raise Exception("Could not get hodring XML-RPC server address.")
#        ringClient = hodXRClient(ringXRAddress, None, None, 0, 0, 0, False)
#        self.log.debug("Invoking clusterStop on " + ringXRAddress)
#        ringThread = func(name='hodring_stop', functionRef=ringClient.clusterStop())
#        ring['thread'] = ringThread
#        ringThread.start()
#
#      for ring in ringList:
#        ringThread = ring['thread']
#        if ringThread == None:
#          raise Exception("Could not get hodring thread.")
#        ringThread.join()
#        self.log.debug("Completed clusterStop on " + ring['xrs'])
#
#    except:
#      self.log.debug(get_exception_string())
#      return False
#
#    self.log.debug("Successfully stopped cluster.")
#    
#    return True

  def getCommand(self, addr):
    """This method is called by the
    hodrings to get commands from
    the ringmaster"""
    lock = self.cmdLock
    cmdList = []
    lock.acquire()
    try:
      try:
        for v in self.serviceDict.itervalues():
          if (not v.isExternal()):
            if v.isLaunchable(self.serviceDict):
              # 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)
                break
        if cmdList == []:
          for s in self.serviceDict.itervalues():
            if (not v.isExternal()):
              if s.isMasterInitialized():
                cl = s.getWorkerCommands(self.serviceDict)
                cmdList.extend(cl)
              else:
                cmdList = []
                break
      except:
        self.log.debug(get_exception_string())
    finally:
      lock.release()
      pass
    
    cmd = addr + pformat(cmdList)
    self.log.debug("getCommand returning " + cmd)
    return cmdList
  
  def getAdminCommand(self, addr):
    """This method is called by the
    hodrings to get admin commands from
    the ringmaster"""
    lock = self.cmdLock
    cmdList = []
    lock.acquire()
    try:
      try:
        for v in self.serviceDict.itervalues():
          cmdList = v.getAdminCommands(self.serviceDict)
          if cmdList != []:
            break
      except Exception, e:
        self.log.debug(get_exception_string())
    finally:
      lock.release()
      pass
    cmd = addr + pformat(cmdList)
    self.log.debug("getAdminCommand returning " + cmd)
    return cmdList

  def addMasterParams(self, addr, vals):
    """This method is called by
    hodring to update any parameters
    its changed for the commands it was
    running"""
    self.log.debug('Comment: adding master params from %s' % addr)
    self.log.debug(pformat(vals))
    lock = self.masterParamLock
    lock.acquire()
    try:
      for v in self.serviceDict.itervalues():
        if v.isMasterLaunched():
          if (v.getMasterAddress() == addr):
            v.setMasterParams(vals)
            v.setMasterInitialized()
    except:
      self.log.debug(get_exception_string())
      pass
    lock.release()
            
    return addr

  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))
    lock = self.masterParamLock
    lock.acquire()
    try:
      for v in self.serviceDict.itervalues():
        if v.isMasterLaunched():
          if (v.getMasterAddress() == addr):
            # strip the PID part.
            idx = addr.rfind('_')
            if idx is not -1:
              addr = addr[:idx]
            v.setMasterFailed("Hodring at %s failed with following" \
                                " errors:\n%s" % (addr, errors))
    except:
      self.log.debug(get_exception_string())
      pass
    lock.release()
    return True

  def getKeys(self):
    lock= self.masterParamLock
    lock.acquire()
    keys = self.serviceDict.keys()
    lock.release()    
  
    return keys
  
  def getServiceAddr(self, name):
    addr = 'not found'
    self.log.debug("getServiceAddr name: %s" % name)
    lock= self.masterParamLock
    lock.acquire()
    try:
      service = self.serviceDict[name]
    except KeyError:
      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) 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]
      else:
        addr = 'not found'
    lock.release()
    self.log.debug("getServiceAddr addr %s: %s" % (name, addr))
    
    return addr

  def getURLs(self, name):
    addr = 'none'
    lock = self.masterParamLock
    lock.acquire()
    
    try:
      service = self.serviceDict[name]
    except KeyError:
      pass
    else:
      if (service.isMasterInitialized()):
        addr = service.getInfoAddrs()[0]
      
    lock.release()
    
    return addr

  def stopRM(self):
    """An XMLRPC call which will spawn a thread to stop the Ringmaster program."""
    # We spawn a thread here because we want the XMLRPC call to return. Calling
    # stop directly from here will also stop the XMLRPC server.
    try:
      self.log.debug("inside xml-rpc call to stop ringmaster")
      rmStopperThread = func('RMStopper', self.rm.stop)
      rmStopperThread.start()
      self.log.debug("returning from xml-rpc call to stop ringmaster")
      return True
    except:
      self.log.debug("Exception in stop: %s" % get_exception_string())
      return False

class RingMaster:
  def __init__(self, cfg, log, **kwds):
    """starts nodepool and services"""
    self.download = False
    self.httpServer = None
    self.cfg = cfg
    self.log = log
    self.__hostname = local_fqdn()
    self.workDirs = None 

    # ref to the idle job tracker object.
    self.__jtMonitor = None
    self.__idlenessDetected = False
    self.__stopInProgress = False
    self.__isStopped = False # to let main exit
    self.__exitCode = 0 # exit code with which the ringmaster main method should return

    self.workers_per_ring = self.cfg['ringmaster']['workers_per_ring']

    self.__initialize_signal_handlers()
    
    sdd = self.cfg['servicedesc']
    gsvc = None
    for key in sdd:
      gsvc = sdd[key]
      break
    
    npd = self.cfg['nodepooldesc']
    self.np = NodePoolUtil.getNodePool(npd, cfg, log)

    self.log.debug("Getting service ID.")
    
    self.serviceId = self.np.getServiceId()
    
    self.log.debug("Got service ID: %s" % self.serviceId)

    self.tarSrcLoc = None
    if self.cfg['ringmaster'].has_key('hadoop-tar-ball'):
      self.download = True
      self.tarSrcLoc = self.cfg['ringmaster']['hadoop-tar-ball']
 
    self.cd_to_tempdir()

    if (self.download):
      self.__copy_tarball(os.getcwd())
      self.basename = self.__find_tarball_in_dir(os.getcwd())
      if self.basename is None:
        raise Exception('Did not find tarball copied from %s in %s.'
                          % (self.tarSrcLoc, os.getcwd()))
      
    self.serviceAddr = to_http_url(self.cfg['ringmaster']['svcrgy-addr'])
    
    self.log.debug("Service registry @ %s" % self.serviceAddr)
    
    self.serviceClient = hodXRClient(self.serviceAddr)
    self.serviceDict  = {}
    try:
      sdl = self.cfg['servicedesc']

      workDirs = self.getWorkDirs(cfg)

      hdfsDesc = sdl['hdfs']
      hdfs = None
 
      # Determine hadoop Version
      hadoopVers = hadoopVersion(self.__getHadoopDir(), \
                                self.cfg['hodring']['java-home'], self.log)
     
      if (hadoopVers['major']==None) or (hadoopVers['minor']==None):
        raise Exception('Could not retrive the version of Hadoop.'
                        + ' Check the Hadoop installation or the value of the hodring.java-home variable.')
      if hdfsDesc.isExternal():
        hdfs = HdfsExternal(hdfsDesc, workDirs, version=int(hadoopVers['minor']))
        hdfs.setMasterParams( self.cfg['gridservice-hdfs'] )
      else:
        hdfs = Hdfs(hdfsDesc, workDirs, 0, version=int(hadoopVers['minor']),
                    workers_per_ring = self.workers_per_ring)

      self.serviceDict[hdfs.getName()] = hdfs
      
      mrDesc = sdl['mapred']
      mr = None
      if mrDesc.isExternal():
        mr = MapReduceExternal(mrDesc, workDirs, version=int(hadoopVers['minor']))
        mr.setMasterParams( self.cfg['gridservice-mapred'] )
      else:
        mr = MapReduce(mrDesc, workDirs,1, version=int(hadoopVers['minor']),
                       workers_per_ring = self.workers_per_ring)

      self.serviceDict[mr.getName()] = mr
    except:
      self.log.critical("Exception in creating Hdfs and Map/Reduce descriptor objects: \
                            %s." % get_exception_error_string())
      self.log.debug(get_exception_string())
      raise

    # should not be starting these in a constructor
    ringMasterServer.startService(self.serviceDict, cfg, self.np, log, self)
    
    self.rpcserver = ringMasterServer.getAddress()
    
    self.httpAddress = None   
    self.tarAddress = None 
    hostname = socket.gethostname()
    if (self.download):
      self.httpServer = threadedHTTPServer(hostname, 
        self.cfg['ringmaster']['http-port-range'])
      
      self.httpServer.serve_forever()
      self.httpAddress = "http://%s:%d/" % (self.httpServer.server_address[0], 
                                 self.httpServer.server_address[1])
      self.tarAddress = "%s%s" % (self.httpAddress, self.basename)
      
      ringMasterServer.instance.logMasterSources.registerTarSource(hostname, 
                                                                   self.tarAddress)
    else:
      self.log.debug("Download not set.")
    
    self.log.debug("%s %s %s %s %s" % (self.cfg['ringmaster']['userid'], 
      self.serviceId, self.__hostname, 'ringmaster', 'hod'))
    
    if self.cfg['ringmaster']['register']:      
      if self.httpAddress:
        self.serviceClient.registerService(self.cfg['ringmaster']['userid'], 
          self.serviceId, self.__hostname, 'ringmaster', 'hod', {
          'xrs' : self.rpcserver, 'http' : self.httpAddress })
      else:
        self.serviceClient.registerService(self.cfg['ringmaster']['userid'], 
          self.serviceId, self.__hostname, 'ringmaster', 'hod', {
          'xrs' : self.rpcserver, })
    
    self.log.debug("Registered with serivce registry: %s." % self.serviceAddr)
    
    hodRingPath = os.path.join(cfg['ringmaster']['base-dir'], 'bin', 'hodring')
    hodRingWorkDir = os.path.join(cfg['hodring']['temp-dir'], 'hodring' + '_' 
                                  + getpass.getuser())
    
    self.cfg['hodring']['hodring'] = [hodRingWorkDir,]
    self.cfg['hodring']['svcrgy-addr'] = self.cfg['ringmaster']['svcrgy-addr']
    self.cfg['hodring']['service-id'] = self.np.getServiceId()

    self.cfg['hodring']['ringmaster-xrs-addr'] = self.__url_to_addr(self.rpcserver)
    
    if (self.tarSrcLoc != None):
      cfg['hodring']['download-addr'] = self.tarAddress
 
    self.__init_job_tracker_monitor(ringMasterServer.instance.logMasterSources)

  def __init_job_tracker_monitor(self, logMasterSources):
    hadoopDir = self.__getHadoopDir()
    self.log.debug('hadoopdir=%s, java-home=%s' % \
                (hadoopDir, self.cfg['hodring']['java-home']))
    try:
      self.__jtMonitor = JobTrackerMonitor(self.log, self, 
                            self.cfg['ringmaster']['jt-poll-interval'], 
                            self.cfg['ringmaster']['idleness-limit'],
                            hadoopDir, self.cfg['hodring']['java-home'],
                            logMasterSources)
      self.log.debug('starting jt monitor')
      self.__jtMonitor.start()
    except:
      self.log.critical('Exception in running idle job tracker. This cluster cannot be deallocated if idle.\
                          Exception message: %s' % get_exception_error_string())
      self.log.debug('Exception details: %s' % get_exception_string())


  def __getHadoopDir(self):
    hadoopDir = None
    if self.cfg['ringmaster'].has_key('hadoop-tar-ball'):
      tarFile = os.path.join(os.getcwd(), self.basename)
      ret = untar(tarFile, os.getcwd())
      if not ret:
        raise Exception('Untarring tarfile %s to directory %s failed. Cannot find hadoop directory.' \
                            % (tarFile, os.getcwd()))
      hadoopDir = os.path.join(os.getcwd(), self.__get_dir(tarFile))
    else:
      hadoopDir = self.cfg['gridservice-mapred']['pkgs']
    self.log.debug('Returning Hadoop directory as: %s' % hadoopDir)
    return hadoopDir

  def __get_dir(self, name):
    """Return the root directory inside the tarball
    specified by name. Assumes that the tarball begins
    with a root directory."""
    import tarfile
    myTarFile = tarfile.open(name)
    hadoopPackage = myTarFile.getnames()[0]
    self.log.debug("tarball name : %s hadoop package name : %s" %(name,hadoopPackage))
    return hadoopPackage

  def __find_tarball_in_dir(self, dir):
    """Find the tarball among files specified in the given 
    directory. We need this method because how the tarball
    source URI is given depends on the method of copy and
    we can't get the tarball name from that.
    This method will fail if there are multiple tarballs
    in the directory with the same suffix."""
    files = os.listdir(dir)
    for file in files:
      if self.tarSrcLoc.endswith(file):
        return file
    return None

  def __copy_tarball(self, destDir):
    """Copy the hadoop tar ball from a remote location to the
    specified destination directory. Based on the URL it executes
    an appropriate copy command. Throws an exception if the command
    returns a non-zero exit code."""
    # for backwards compatibility, treat the default case as file://
    url = ''
    if self.tarSrcLoc.startswith('/'):
      url = 'file:/'
    src = '%s%s' % (url, self.tarSrcLoc)
    if src.startswith('file://'):
      src = src[len('file://')-1:]
      cpCmd = '/bin/cp'
      cmd = '%s %s %s' % (cpCmd, src, destDir)
      self.log.debug('Command to execute: %s' % cmd)
      copyProc = simpleCommand('remote copy', cmd)
      copyProc.start()
      copyProc.wait()
      copyProc.join()
      ret = copyProc.exit_code()
      self.log.debug('Completed command execution. Exit Code: %s.' % ret)

      if ret != 0:
        output = copyProc.output()
        raise Exception('Could not copy tarball using command %s. Exit code: %s. Output: %s' 
                        % (cmd, ret, output))
    else:
      raise Exception('Unsupported URL for file: %s' % src)

# input: http://hostname:port/. output: [hostname,port]
  def __url_to_addr(self, url):
    addr = url.rstrip('/')
    if addr.startswith('http://'):
      addr = addr.replace('http://', '', 1)
    addr_parts = addr.split(':')
    return [addr_parts[0], int(addr_parts[1])]

  def __initialize_signal_handlers(self): 
    def sigStop(sigNum, handler):
      sig_wrapper(sigNum, self.stop)
  
    signal.signal(signal.SIGTERM, sigStop)
    signal.signal(signal.SIGINT, sigStop)
    signal.signal(signal.SIGQUIT, sigStop)

  def __clean_up(self):
    tempDir = self.__get_tempdir()
    os.chdir(os.path.split(tempDir)[0])
    if os.path.exists(tempDir):
      shutil.rmtree(tempDir, True)
      
    self.log.debug("Cleaned up temporary dir: %s" % tempDir)

  def __get_tempdir(self):
    dir = os.path.join(self.cfg['ringmaster']['temp-dir'], 
                          "%s.%s.ringmaster" % (self.cfg['ringmaster']['userid'], 
                                                self.np.getServiceId()))
    return dir

  def getWorkDirs(self, cfg, reUse=False):

    if (not reUse) or (self.workDirs == None):
      import math
      frand = random.random()
      while math.ceil(frand) != math.floor(frand):
        frand = frand * 100

      irand = int(frand)
      uniq = '%s-%d-%s' % (socket.gethostname(), os.getpid(), irand)
      dirs = []
      parentDirs = cfg['ringmaster']['work-dirs']
      for p in parentDirs:
        dir = os.path.join(p, uniq)
        dirs.append(dir)
      self.workDirs = dirs

    return self.workDirs

  def _fetchLink(self, link, parentDir):
    parser = miniHTMLParser()
    self.log.debug("Checking link %s" %link)
    while link:

      # Get the file from the site and link
      input = urllib.urlopen(link)
      out = None
      contentType = input.info().gettype()
      isHtml = contentType == 'text/html'

      #print contentType
      if isHtml:
        parser.setBaseUrl(input.geturl())
      else:
        parsed = urlparse.urlparse(link)
        hp = parsed[1]
        h = hp
        p = None
        if hp.find(':') != -1:
          h, p = hp.split(':', 1)
        path = parsed[2]
        path = path.split('/')
        file = os.path.join(parentDir, h, p)
        for c in path:
          if c == '':
            continue
          file = os.path.join(file, c)

        try:
          self.log.debug('Creating %s' % file)
          dir, tail = os.path.split(file)
          if not os.path.exists(dir):
            os.makedirs(dir)
        except:
          self.log.debug(get_exception_string())

        out = open(file, 'w')

      bufSz = 8192
      buf = input.read(bufSz)
      while len(buf) > 0:
        if isHtml:
          # Feed the file into the HTML parser
          parser.feed(buf)
        if out:
          out.write(buf)
        buf = input.read(bufSz)

      input.close()
      if out:
        out.close()

      # Search the retfile here

      # Get the next link in level traversal order
      link = parser.getNextLink()
      
    parser.close()
    
  def _finalize(self):
    try:
      # FIXME: get dir from config
      dir = 'HOD-log-P%d' % (os.getpid())
      dir = os.path.join('.', dir)
    except:
      self.log.debug(get_exception_string())

    self.np.finalize()

  def handleIdleJobTracker(self):
    self.log.critical("Detected idle job tracker for %s seconds. The allocation will be cleaned up." \
                          % self.cfg['ringmaster']['idleness-limit'])
    self.__idlenessDetected = True

  def cd_to_tempdir(self):
    dir = self.__get_tempdir()
    
    if not os.path.exists(dir):
      os.makedirs(dir)
    os.chdir(dir)
    
    return dir
  
  def getWorkload(self):
    return self.workload

  def getHostName(self):
    return self.__hostname

  def start(self):
    """run the thread main loop"""
    
    self.log.debug("Entered start method.")
    hodring = os.path.join(self.cfg['ringmaster']['base-dir'], 
                           'bin', 'hodring')
    largs = [hodring]
    targs = self.cfg.get_args(section='hodring')
    largs.extend(targs) 
    
    hodringCmd = ""
    for item in largs:
      hodringCmd = "%s%s " % (hodringCmd, item)
      
    self.log.debug(hodringCmd)
    
    if self.np.runWorkers(largs) > 0:
      self.log.critical("Failed to start worker.")
    
    self.log.debug("Returned from runWorkers.")
    
    self._finalize()

  def __findExitCode(self):
    """Determine the exit code based on the status of the cluster or jobs run on them"""
    xmlrpcServer = ringMasterServer.instance.logMasterSources
    if xmlrpcServer.getServiceAddr('hdfs') == 'not found' or \
        xmlrpcServer.getServiceAddr('hdfs').startswith("Error: "):
      self.__exitCode = 7
    elif xmlrpcServer.getServiceAddr('mapred') == 'not found' or \
        xmlrpcServer.getServiceAddr('mapred').startswith("Error: "):
      self.__exitCode = 8
    else:
      clusterStatus = get_cluster_status(xmlrpcServer.getServiceAddr('hdfs'),
                                          xmlrpcServer.getServiceAddr('mapred'))
      if clusterStatus != 0:
        self.__exitCode = clusterStatus
      else:
        self.__exitCode = self.__findHadoopJobsExitCode()
    self.log.debug('exit code %s' % self.__exitCode)

  def __findHadoopJobsExitCode(self):
    """Determine the consolidate exit code of hadoop jobs run on this cluster, provided
       this information is available. Return 0 otherwise"""
    ret = 0
    failureStatus = 3
    failureCount = 0
    if self.__jtMonitor:
      jobStatusList = self.__jtMonitor.getJobsStatus()
      try:
        if len(jobStatusList) > 0:
          for jobStatus in jobStatusList:
            self.log.debug('job status for %s: %s' % (jobStatus.getJobId(), 
                                                      jobStatus.getStatus()))
            if jobStatus.getStatus() == failureStatus:
              failureCount = failureCount+1
        if failureCount > 0:
          if failureCount == len(jobStatusList): # all jobs failed
            ret = 16
          else:
            ret = 17
      except:
        self.log.debug('exception in finding hadoop jobs exit code' % get_exception_string())
    return ret

  def stop(self):
    self.log.debug("RingMaster stop method invoked.")
    if self.__stopInProgress or self.__isStopped:
      return
    self.__stopInProgress = True
    if ringMasterServer.instance is not None:
      self.log.debug('finding exit code')
      self.__findExitCode()
      self.log.debug('stopping ringmaster instance')
      ringMasterServer.stopService()
    else:
      self.__exitCode = 6
    if self.__jtMonitor is not None:
      self.__jtMonitor.stop()
    if self.httpServer:
      self.httpServer.stop()
      
    self.__clean_up()
    self.__isStopped = True

  def shouldStop(self):
    """Indicates whether the main loop should exit, either due to idleness condition, 
    or a stop signal was received"""
    return self.__idlenessDetected or self.__isStopped

  def getExitCode(self):
    """return the exit code of the program"""
    return self.__exitCode

def main(cfg,log):
  try:
    rm = None
    dGen = DescGenerator(cfg)
    cfg = dGen.initializeDesc()
    rm = RingMaster(cfg, log)
    rm.start()
    while not rm.shouldStop():
      time.sleep(1)
    rm.stop()
    log.debug('returning from main')
    return rm.getExitCode()
  except Exception, e:
    if log:
      log.critical(get_exception_string())
    raise Exception(e)