hod

HADOOP 0.18.0安装源代码头文件

             
             ('cmdline-params', 'keyval', 'Hadoop cmdline key/value list.',
              False, None, False, False),

             ('server-params', 'keyval', 'Hadoop xml key/value list',
              False, None, False, True, 'H'),

             ('final-server-params', 'keyval', 'Hadoop final xml key/value list',
              False, None, False, True, 'S'),
           
             ('envs', 'keyval', 'Environment in which to run this package.',
              False, None, False, True),

             ('pkgs', 'directory', "directory where the package is installed",
              False, None, False, False)),           
             
             
            'hodring' : (
             ('temp-dir', 'list', 'hodring temporary directory.',
              False, None, True, False),
              
             ('log-dir', 'directory', 'hod logging directory.', 
              False, os.path.join(rootDirectory, 'logs'), False, False), 
              
             ('log-destination-uri', 'string', 
              'URI to store logs to, local://some_path or '
              + 'hdfs://host:port/some_path', 
              False, None, False, True),

             ('pkgs', 'directory', 'Path to Hadoop to use in case of uploading to HDFS',
              False, None, False, False),
              
             ('syslog-address', 'address', 'Syslog address.',
              False, None, False, True),
          
             ('java-home', 'directory', 'Java home directory.',
              False, None, True, False),
              
             ('debug', 'pos_int', 'Debugging level, 0-4.',
              False, 3, True, True),
               
             ('register', 'bool', 'Register with service registry?',
              False, True, True, True),
               
             ('stream', 'bool', 'Output to stderr.',
              False, False, False, True),

             ('userid', 'user_account', 
              'User ID the hod shell is running under.',
              False, pwd.getpwuid(os.getuid())[0], False, True),
               
             ('command', 'string', 'Command for hodring to run.',
              False, None, False, True),

             ('xrs-port-range', 'range', 'XML-RPC port range n-m.',
              False, None, True, True),
               
             ('http-port-range', 'range', 'HTTP port range n-m.',
              False, None, True, True),
              
             ('service-id', 'string', 'Service ID.',
              False, None, False, True),
              
             ('download-addr', 'string', 'Download HTTP address.',
              False, None, False, True),
               
             ('svcrgy-addr', 'address', 'Download HTTP address.',
              False, None, False, True), 
    
             ('ringmaster-xrs-addr', 'address', 'Ringmaster XML-RPC address.',
              False, None, False, True),

             ('tarball-retry-initial-time', 'pos_float','Initial Retry time for tarball download',
              False, 1, False, True),
              
             ('tarball-retry-interval', 'pos_float','interval to spread retries for tarball download',
              False, 3, False, True),
              
             ('cmd-retry-initial-time', 'pos_float','Initial retry time for getting commands',
              False, 2, False, True),
             
             ('cmd-retry-interval', 'pos_float','interval to spread retries for getting commands',
              False, 2, False, True),

             ('mapred-system-dir-root', 'string', 'Root under which mapreduce system directory names are generated by HOD.',
              False, '/mapredsystem', False, False))
              }   

defOrder = [ 'hod', 'ringmaster', 'hodring', 'resource_manager', 
             'gridservice-mapred', 'gridservice-hdfs' ]

def printErrors(msgs):
  for msg in msgs:
    print msg

def op_requires_pkgs(config):
  if config['hod'].has_key('operation'):
    return config['hod']['operation'].startswith('allocate')
  else:
    return config['hod'].has_key('script')

if __name__ == '__main__':  
  try:
    confDef = definition()
    confDef.add_defs(defList, defOrder)
    hodhelp = hodHelp()
    usage = hodhelp.help()
            
    hodOptions = options(confDef, usage,
                      VERSION, withConfig=True, defaultConfig=DEFAULT_CONFIG,
                      name=myName )
    # hodConfig is a dict like object, hodConfig[section][name]
    try:
      hodConfig = config(hodOptions['config'], configDef=confDef, 
                       originalDir=hodOptions['hod']['original-dir'],
                       options=hodOptions) 
    except IOError, e:
      print >>sys.stderr,"error: %s not found. Specify the path to the HOD configuration file, or define the environment variable %s under which a file named hodrc can be found." % (hodOptions['config'], 'HOD_CONF_DIR')
      sys.exit(1)
  
    # Conditional validation
    statusMsgs = []

    if hodConfig.normalizeValue('gridservice-hdfs', 'external'):
      # For external HDFS
      statusMsgs.extend(hodConfig.validateValue('gridservice-hdfs',
                                                'fs_port'))
      statusMsgs.extend(hodConfig.validateValue('gridservice-hdfs',
                                                'info_port'))
      statusMsgs.extend(hodConfig.validateValue('gridservice-hdfs',
                                                'host'))
    else:
      hodConfig['gridservice-hdfs']['fs_port'] = 0 # Dummy
      hodConfig['gridservice-hdfs']['info_port'] = 0 # Not used at all

    if hodConfig.normalizeValue('gridservice-mapred', 'external'):
      statusMsgs.extend(hodConfig.validateValue('gridservice-mapred',
                                                'tracker_port'))
      statusMsgs.extend(hodConfig.validateValue('gridservice-mapred',
                                                'info_port'))
      statusMsgs.extend(hodConfig.validateValue('gridservice-mapred',
                                                'host'))
    else:
      hodConfig['gridservice-mapred']['tracker_port'] = 0 # Dummy
      hodConfig['gridservice-mapred']['info_port'] = 0 # Not used at all

    if len(statusMsgs) != 0:
      for msg in statusMsgs:
        print >>sys.stderr, msg
      sys.exit(1)
    # End of conditional validation

    status = True
    statusMsgs = []
  
    (status,statusMsgs) = hodConfig.verify()
    if not status:
      print >>sys.stderr,"error: bin/hod failed to start."
      for msg in statusMsgs:
        print >>sys.stderr,"%s" % (msg)
      sys.exit(1)
  
    ## TODO : should move the dependency verification to hodConfig.verify
    if hodConfig['hod'].has_key('operation') and \
      hodConfig['hod'].has_key('script'):
      print "Script operation is mutually exclusive with other HOD operations"
      hodOptions.print_help(sys.stderr)
      sys.exit(1)
    
    if 'operation' not in hodConfig['hod'] and 'script' not in hodConfig['hod']:
      print "HOD requires at least a script or operation be specified."
      hodOptions.print_help(sys.stderr)
      sys.exit(1)    
    
    if hodConfig['gridservice-hdfs']['external']:
      hdfsAddress = "%s:%s" % (hodConfig['gridservice-hdfs']['host'], 
                               hodConfig['gridservice-hdfs']['fs_port'])
  
      hdfsSocket = tcpSocket(hdfsAddress)
        
      try:
        hdfsSocket.open()
        hdfsSocket.close()
      except tcpError:
        printErrors(hodConfig.var_error('hod', 'gridservice-hdfs', 
          "Failed to open a connection to external hdfs address: %s." % 
          hdfsAddress))
        sys.exit(1)
    else:
      hodConfig['gridservice-hdfs']['host'] = 'localhost'
  
    if hodConfig['gridservice-mapred']['external']:
      mapredAddress = "%s:%s" % (hodConfig['gridservice-mapred']['host'], 
                                 hodConfig['gridservice-mapred']['tracker_port'])
  
      mapredSocket = tcpSocket(mapredAddress)
        
      try:
        mapredSocket.open()
        mapredSocket.close()
      except tcpError:
        printErrors(hodConfig.var_error('hod', 'gridservice-mapred', 
          "Failed to open a connection to external mapred address: %s." % 
          mapredAddress))
        sys.exit(1)
    else:
      hodConfig['gridservice-mapred']['host'] = 'localhost'
  
    if not hodConfig['ringmaster'].has_key('hadoop-tar-ball') and \
      not hodConfig['gridservice-hdfs'].has_key('pkgs') and \
      op_requires_pkgs(hodConfig):
      printErrors(hodConfig.var_error('gridservice-hdfs', 'pkgs', 
        "gridservice-hdfs.pkgs must be defined if ringmaster.hadoop-tar-ball "
        + "is not defined."))
      sys.exit(1)
  
    if not hodConfig['ringmaster'].has_key('hadoop-tar-ball') and \
      not hodConfig['gridservice-mapred'].has_key('pkgs') and \
      op_requires_pkgs(hodConfig):
      printErrors(hodConfig.var_error('gridservice-mapred', 'pkgs', 
        "gridservice-mapred.pkgs must be defined if ringmaster.hadoop-tar-ball "
        + "is not defined."))
      sys.exit(1)
  
    if hodConfig['hodring'].has_key('log-destination-uri'):
      if hodConfig['hodring']['log-destination-uri'].startswith('file://'):
        pass
      elif hodConfig['hodring']['log-destination-uri'].startswith('hdfs://'):
        hostPort = hodConfig['hodring']['log-destination-uri'][7:].split("/")
        hostPort = hostPort[0]
        socket = tcpSocket(hostPort)
        try:
          socket.open()
          socket.close()
        except:
          printErrors(hodConfig.var_error('hodring', 'log-destination-uri', 
          "Unable to contact host/port specified in log destination uri: %s" % 
          hodConfig['hodring']['log-destination-uri']))
          sys.exit(1)
      else:
        printErrors(hodConfig.var_error('hodring', 'log-destination-uri', 
          "The log destiniation uri must be of type local:// or hdfs://."))
        sys.exit(1)
  
    ## TODO : end of should move the dependency verification to hodConfig.verif
      
    hodConfig['hod']['base-dir'] = rootDirectory
    hodConfig['hod']['user_state'] = DEFAULT_HOD_DIR
  
    dGen = DescGenerator(hodConfig)
    hodConfig = dGen.initializeDesc()
    
    os.environ['JAVA_HOME'] = hodConfig['hod']['java-home']
    
    if hodConfig['hod']['debug'] == 4:
      print ""
      print "Using Python: %s" % sys.version
      print ""
   
    hod = hodRunner(hodConfig)
  
    # Initiate signal handling
    hodInterrupt.set_log(hod.get_logger())
    hodInterrupt.init_signals()
    # Interrupts set up. Now on we handle signals only when we wish to.
  except KeyboardInterrupt:
    print HOD_INTERRUPTED_MESG
    sys.exit(HOD_INTERRUPTED_CODE)
  
  if hodConfig['hod'].has_key('script'):
    sys.exit(hod.script())
  else:  
    sys.exit(hod.operation())