common/src/contrib/hod/bin/hodcleanup

#!/bin/sh

# 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.


""":"
work_dir=$(dirname $0)
base_name=$(basename $0)
original_dir=$PWD
cd $work_dir

if [ $HOD_PYTHON_HOME ]; then
    exec $HOD_PYTHON_HOME -u -OO $base_name ${1+"$@"}
elif [ -e /usr/bin/python ]; then
    exec /usr/bin/python -u -OO $base_name ${1+"$@"}
elif [ -e /usr/local/bin/python ]; then
    exec /usr/local/bin/python -u -OO $base_name ${1+"$@"}
else
    exec python -u -OO $base_name ${1+"$@"}
fi
":"""

"""The executable to be used by the user"""
import sys, os, re, pwd, threading, sys, random, time, pprint, shutil, time, re
from pprint import pformat
from optparse import OptionParser

myName          = os.path.basename(sys.argv[0])
myName          = re.sub(".*/", "", myName)
binDirectory    = os.path.realpath(sys.argv[0])
rootDirectory   = re.sub("/bin/.*", "", binDirectory)
libDirectory    = rootDirectory

sys.path.append(libDirectory)

from hodlib.Common.threads import simpleCommand
from hodlib.Common.util import local_fqdn, tar, filter_warnings,\
                            get_exception_string, get_exception_error_string
from hodlib.Common.logger import hodLog
from hodlib.Common.logger import getLogger
from hodlib.HodRing.hodRing import createMRSystemDirectoryManager

filter_warnings()

reVersion = re.compile(".*(\d+_\d+).*")
reHdfsURI = re.compile("(hdfs://.*?:\d+)(.*)")

VERSION = None
if os.path.exists("./VERSION"):
  vFile = open("./VERSION", 'r')
  VERSION = vFile.readline()
  vFile.close()

def __archive_logs(conf, log):
  # need log-destination-uri, __hadoopLogDirs, temp-dir
  status = True
  logUri = conf['log-destination-uri']
  hadoopLogDirs = conf['hadoop-log-dirs']
  if logUri:
    try:
      if hadoopLogDirs:
        date = time.localtime()
        for logDir in hadoopLogDirs:
          (head, tail) = os.path.split(logDir)
          (head, logType) = os.path.split(head)
          tarBallFile = "%s-%s-%04d%02d%02d%02d%02d%02d-%s.tar.gz" % (
            logType, local_fqdn(), date[0], date[1], date[2], date[3], 
            date[4], date[5], random.randint(0,1000))
          
          if logUri.startswith('file://'):
            tarBallFile = os.path.join(logUri[7:], 
                                       tarBallFile)
          else:
            tarBallFile = os.path.join(conf['temp-dir'], tarBallFile)
          
          log.debug('archiving log files to: %s' % tarBallFile)
          status = tar(tarBallFile, logDir, ['*',])
          log.info('archive %s status: %s' % (tarBallFile, status))
          if status and \
            logUri.startswith('hdfs://'):
            __copy_archive_to_dfs(conf, tarBallFile)
            log.info("copying archive to dfs finished")
        dict = {} 
    except:
      log.error(get_exception_string())
      status = False
  return status


def __copy_archive_to_dfs(conf, archiveFile):
  # need log-destination-uri, hadoopCommandstring and/or pkgs
  hdfsURIMatch = reHdfsURI.match(conf['log-destination-uri'])
  
  (head, tail) = os.path.split(archiveFile)
  destFile = os.path.join(hdfsURIMatch.group(2), conf['user-id'], 'hod-logs', conf['service-id'], tail)
  
  log.info("copying archive %s to DFS %s ..." % (archiveFile, destFile))
  
  hadoopCmd = conf['hadoop-command-string']
  if conf['pkgs']:
    hadoopCmd = os.path.join(conf['pkgs'], 'bin', 'hadoop')

  copyCommand = "%s dfs -fs %s -copyFromLocal %s %s" % (hadoopCmd, 
    hdfsURIMatch.group(1), archiveFile, destFile)
  
  log.debug(copyCommand)
  
  copyThread = simpleCommand('hadoop', copyCommand)
  copyThread.start()
  copyThread.wait()
  copyThread.join()
  log.debug(pprint.pformat(copyThread.output()))
  
  os.unlink(archiveFile)

def unpack():
  parser = OptionParser()
  option_list=["--log-destination-uri", "--hadoop-log-dirs", \
          "--temp-dir", "--hadoop-command-string", "--pkgs", "--user-id", \
          "--service-id", "--hodring-debug", "--hodring-log-dir", \
          "--hodring-syslog-address", "--hodring-cleanup-list", \
          "--jt-pid", "--mr-sys-dir", "--fs-name", "--hadoop-path"]
  regexp = re.compile("^--")
  for opt in option_list:
    parser.add_option(opt,dest=regexp.sub("",opt),action="store")
  option_list.append("--hodring-stream")
  parser.add_option("--hodring-stream",dest="hodring-stream",metavar="bool",\
                                                        action="store_true")
  (options, args) = parser.parse_args()
  _options= {}
  _options['hodring'] = {}
  for opt in dir(options):
    if "--"+opt in option_list:
      _options[opt] = getattr(options,opt)
  if _options.has_key('hadoop-log-dirs') and _options['hadoop-log-dirs']:
    _options['hadoop-log-dirs'] = _options['hadoop-log-dirs'].split(",")
  if _options.has_key('hodring-syslog-address') and _options['hodring-syslog-address']:
    _options['hodring']['syslog-address'] = \
        _options['hodring-syslog-address'].split(':')
  _options['hodring']['debug']        = int(_options['hodring-debug'])
  _options['hodring']['log-dir']      = _options['hodring-log-dir']
  _options['hodring']['stream']      = _options['hodring-stream']
  _options['hodring']['userid']      = _options['user-id']
  os.putenv('PBS_JOBID', _options['service-id'] )
  return _options
 
if __name__ == '__main__':  
  log = None
  try:
    conf = unpack()
    # Use the same log as hodring
    log = getLogger(conf['hodring'],'hodring')
    log.debug("Logger initialised successfully")
    mrSysDirManager = createMRSystemDirectoryManager(conf, log)
    if mrSysDirManager is not None:
      mrSysDirManager.removeMRSystemDirectory()

    status =  __archive_logs(conf,log)
    log.info("Archive status : %s" % status)
    list = conf['hodring-cleanup-list'].split(',')
    log.info("now removing %s" % list)
    for dir in list:
     if os.path.exists(dir):
       log.debug('removing %s' % (dir))
       shutil.rmtree(dir, True)
       log.debug("done")
    log.info("Cleanup successfully completed")
  except Exception, e:
    if log:
      log.info("Stack trace:\n%s\n%s" %(get_exception_error_string(),get_exception_string()))