hadoop/src/contrib/hod/hodlib/GridServices/mapred.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.
"""define MapReduce as subclass of Service"""

# -*- python -*-

import os, copy, time

from service import *
from hodlib.Hod.nodePool import *
from hodlib.Common.desc import CommandDesc
from hodlib.Common.util import get_exception_string, parseEquals

class MapReduceExternal(MasterSlave):
  """dummy proxy to external MapReduce instance"""

  def __init__(self, serviceDesc, workDirs, version):
    MasterSlave.__init__(self, serviceDesc, workDirs,None)
    self.launchedMaster = True
    self.masterInitialized = True
    self.version = version
    
  def getMasterRequest(self):
    return None

  def getMasterCommands(self, serviceDict):
    return []

  def getAdminCommands(self, serviceDict):
    return []

  def getWorkerCommands(self, serviceDict):
    return []

  def getMasterAddrs(self):
    attrs = self.serviceDesc.getfinalAttrs()
    addr = attrs['mapred.job.tracker']
    return [addr]

  def needsMore(self):
    return 0

  def needsLess(self):
    return 0

  def setMasterParams(self, dict):
    self.serviceDesc['final-attrs']['mapred.job.tracker'] = "%s:%s" % (dict['host'], 
      dict['tracker_port'])
    
    if self.version < 16:
      self.serviceDesc.dict['final-attrs']['mapred.job.tracker.info.port'] = \
                                      str(self.serviceDesc.dict['info_port'])
    else:
      # After Hadoop-2185
      self.serviceDesc['final-attrs']['mapred.job.tracker.http.address'] = \
        "%s:%s" %(dict['host'], dict['info_port'])

  def getInfoAddrs(self):
    attrs = self.serviceDesc.getfinalAttrs()
    if self.version < 16:
      addr = attrs['mapred.job.tracker']
      k,v = addr.split( ":")
      infoaddr = k + ':' + attrs['mapred.job.tracker.info.port']
    else:
      # After Hadoop-2185
      # Note: earlier,we never respected mapred.job.tracker.http.address
      infoaddr = attrs['mapred.job.tracker.http.address']
    return [infoaddr]
  
class MapReduce(MasterSlave):

  def __init__(self, serviceDesc, workDirs,required_node, version,
                workers_per_ring = 1):
    MasterSlave.__init__(self, serviceDesc, workDirs,required_node)

    self.masterNode = None
    self.masterAddr = None
    self.infoAddr = None
    self.workers = []
    self.required_node = required_node
    self.version = version
    self.workers_per_ring = workers_per_ring

  def isLaunchable(self, serviceDict):
    hdfs = serviceDict['hdfs']
    if (hdfs.isMasterInitialized()):
      return True
    return False
  
  def getMasterRequest(self):
    req = NodeRequest(1, [], False)
    return req

  def getMasterCommands(self, serviceDict):

    hdfs = serviceDict['hdfs']

    cmdDesc = self._getJobTrackerCommand(hdfs)
    return [cmdDesc]

  def getAdminCommands(self, serviceDict):
    return []

  def getWorkerCommands(self, serviceDict):

    hdfs = serviceDict['hdfs']

    workerCmds = []
    for id in range(1, self.workers_per_ring + 1):
      workerCmds.append(self._getTaskTrackerCommand(str(id), hdfs))
      
    return workerCmds

  def setMasterNodes(self, list):
    node = list[0]
    self.masterNode = node

  def getMasterAddrs(self):
    return [self.masterAddr]

  def getInfoAddrs(self):
    return [self.infoAddr]

  def getWorkers(self):
    return self.workers

  def requiredNode(self):
    return self.required_host

  def setMasterParams(self, list):
    dict = self._parseEquals(list)
    self.masterAddr = dict['mapred.job.tracker']
    k,v = self.masterAddr.split(":")
    self.masterNode = k
    if self.version < 16:
      self.infoAddr = self.masterNode + ':' + dict['mapred.job.tracker.info.port']
    else:
      # After Hadoop-2185
      self.infoAddr = dict['mapred.job.tracker.http.address']
  
  def _parseEquals(self, list):
    return parseEquals(list)

  def _setWorkDirs(self, workDirs, envs, attrs, parentDirs, subDir):
    local = []
    system = None
    temp = None
    hadooptmpdir = None
    dfsclient = []
    
    for p in parentDirs:
      workDirs.append(p)
      workDirs.append(os.path.join(p, subDir))
      dir = os.path.join(p, subDir, 'mapred-local')
      local.append(dir)
      if not system:
        system = os.path.join(p, subDir, 'mapred-system')
      if not temp:
        temp = os.path.join(p, subDir, 'mapred-temp')
      if not hadooptmpdir:
        # Not used currently, generating hadooptmpdir just in case
        hadooptmpdir = os.path.join(p, subDir, 'hadoop-tmp')
      dfsclientdir = os.path.join(p, subDir, 'dfs-client')
      dfsclient.append(dfsclientdir)
      workDirs.append(dfsclientdir)
    # FIXME!! use csv
    attrs['mapred.local.dir'] = ','.join(local)
    attrs['mapred.system.dir'] = 'fillindir'
    attrs['mapred.temp.dir'] = temp
    attrs['dfs.client.buffer.dir'] = ','.join(dfsclient)
    attrs['hadoop.tmp.dir'] = hadooptmpdir


    envs['HADOOP_ROOT_LOGGER'] = "INFO,DRFA"


  def _getJobTrackerCommand(self, hdfs):
    sd = self.serviceDesc

    parentDirs = self.workDirs
    workDirs = []
    attrs = sd.getfinalAttrs().copy()
    envs = sd.getEnvs().copy()

    if 'mapred.job.tracker' not in attrs:
      attrs['mapred.job.tracker'] = 'fillinhostport'

    if self.version < 16:
      if 'mapred.job.tracker.info.port' not in attrs:
        attrs['mapred.job.tracker.info.port'] = 'fillinport'
    else:
      # Addressing Hadoop-2185,
      if 'mapred.job.tracker.http.address' not in attrs:
        attrs['mapred.job.tracker.http.address'] = 'fillinhostport'

    attrs['fs.default.name'] = hdfs.getMasterAddrs()[0]

    self._setWorkDirs(workDirs, envs, attrs, parentDirs, 'mapred-jt')

    dict = { 'name' : 'jobtracker' }
    dict['version'] = self.version
    dict['program'] = os.path.join('bin', 'hadoop')
    dict['argv'] = ['jobtracker']
    dict['envs'] = envs
    dict['pkgdirs'] = sd.getPkgDirs()
    dict['workdirs'] = workDirs
    dict['final-attrs'] = attrs
    dict['attrs'] = sd.getAttrs()
    cmd = CommandDesc(dict)
    return cmd

  def _getTaskTrackerCommand(self, id, hdfs):

    sd = self.serviceDesc

    parentDirs = self.workDirs
    workDirs = []
    attrs = sd.getfinalAttrs().copy()
    envs = sd.getEnvs().copy()
    jt = self.masterAddr

    if jt == None:
      raise ValueError, "Can't get job tracker address"

    attrs['mapred.job.tracker'] = jt
    attrs['fs.default.name'] = hdfs.getMasterAddrs()[0]

    if self.version < 16:
      if 'tasktracker.http.port' not in attrs:
        attrs['tasktracker.http.port'] = 'fillinport'
      # earlier to 16, tasktrackers always took ephemeral port 0 for
      # tasktracker.report.bindAddress
    else:
      # Adding the following. Hadoop-2185
      if 'mapred.task.tracker.report.address' not in attrs:
        attrs['mapred.task.tracker.report.address'] = 'fillinhostport'
      if 'mapred.task.tracker.http.address' not in attrs:
        attrs['mapred.task.tracker.http.address'] = 'fillinhostport'

    # unique parentDirs in case of multiple tasktrackers per hodring
    pd = []
    for dir in parentDirs:
      dir = dir + "-" + id
      pd.append(dir)
    parentDirs = pd
    # end of unique workdirs

    self._setWorkDirs(workDirs, envs, attrs, parentDirs, 'mapred-tt')

    dict = { 'name' : 'tasktracker' }
    dict['program'] = os.path.join('bin', 'hadoop')
    dict['argv'] = ['tasktracker']
    dict['envs'] = envs
    dict['pkgdirs'] = sd.getPkgDirs()
    dict['workdirs'] = workDirs
    dict['final-attrs'] = attrs
    dict['attrs'] = sd.getAttrs()
    cmd = CommandDesc(dict)
    return cmd