#!/usr/bin/env python

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

import logging
import logging.handlers
from optparse import OptionParser
import sys
import os
import ConfigParser
import errno
import shutil
import shlex
import subprocess
import time
import fnmatch
from SliderTesterConfig import SliderTesterConfig

logger = logging.getLogger()
formatstr = "%(levelname)s %(asctime)s %(filename)s:%(lineno)d - %(message)s"


class SliderTester:
  SLIDER_CREATE_CMD_FORMAT = '{0} create {1} --image {2}{3} --option agent.conf {2}{4}  --template {5} --resources {6}  --option application.def {2}{7}'
  HDFS_LS = "hdfs dfs -ls"
  HDFS_CHOWN = "hdfs dfs -chown"
  HDFS_MKDIR = "hdfs dfs -mkdir"
  HDFS_RMR = "hdfs dfs -rm -r"

  def __init__(self, config):
    self.config = config
    self.slider_user = self.config.get(SliderTesterConfig.TEST_SECTION, SliderTesterConfig.APP_USER)
    self.hdfs_user = self.config.get(SliderTesterConfig.TEST_SECTION, SliderTesterConfig.HDFS_ROOT_USER)
    self.jdk_path = self.config.get(SliderTesterConfig.SLIDER_SECTION, SliderTesterConfig.JDK_PATH)

  def run_slider_command(self, cmd, user=None):
    if self.jdk_path and len(self.jdk_path) > 0:
      cmd = "PATH=$PATH:{0}&&{1}".format(self.jdk_path, cmd)

    return self.run_os_command(cmd, user)

  def run_os_command(self, cmd, user=None):
    FORMAT_WITH_SU = 'su -l {0} -c "{1}"'
    if user:
      cmd = FORMAT_WITH_SU.format(user, cmd)

    logger.info('Executing command: ' + str(cmd))
    if type(cmd) == str:
      cmd = shlex.split(cmd)

    process = subprocess.Popen(cmd,
                               stdout=subprocess.PIPE,
                               stdin=subprocess.PIPE,
                               stderr=subprocess.PIPE
    )
    (stdoutdata, stderrdata) = process.communicate()

    logger.debug("Out: " + stdoutdata)
    logger.debug("Err: " + stderrdata)
    logger.debug("Process ret code: " + str(process.returncode))
    return process.returncode, stdoutdata, stderrdata

  def ensure_path_exists(self, path):
    try:
      os.makedirs(path)
    except OSError as exception:
      if exception.errno != errno.EEXIST:
        raise
      pass
    pass

  def ensure_ownership(self, path, user):
    CMD = "chown -R {0} {1}"
    self.run_os_command(CMD.format(user, path))
    pass

  def deploy_and_validate_slider(self):
    UNTAR = "tar -xvf {0} -C {1}"
    self.slider_package = self.config.get(SliderTesterConfig.SLIDER_SECTION, SliderTesterConfig.PACKAGE)
    if not os.path.isfile(self.slider_package):
      raise Exception("Invalid Slider package: " + self.slider_package)

    self.app_package = self.config.get(SliderTesterConfig.APP_SECTION, SliderTesterConfig.PACKAGE)
    if not os.path.isfile(self.app_package):
      raise Exception("Invalid App package: " + self.app_package)

    self.test_root = self.config.get(SliderTesterConfig.TEST_SECTION, SliderTesterConfig.TEST_ROOT)
    if os.path.exists(self.test_root):
      if os.path.isfile(self.test_root):
        os.remove(self.test_root)
      else:
        shutil.rmtree(self.test_root)
    pass

    self.slider_root = os.path.join(self.test_root, "slider")
    self.ensure_path_exists(self.slider_root)
    self.ensure_ownership(self.slider_root, self.slider_user)

    self.run_os_command(UNTAR.format(self.slider_package, self.slider_root), self.slider_user)

    self.app_root = os.path.join(self.test_root, "expandedapp")
    self.ensure_path_exists(self.app_root)
    self.ensure_ownership(self.app_root, self.slider_user)

    self.run_os_command(UNTAR.format(self.app_package, self.app_root), self.slider_user)

    self.app_resources = self.findfile("resources.json", self.app_root)
    self.app_conf = self.findfile("appConfig.json", self.app_root)

    self.slider_exec = self.findfile("slider", self.slider_root)

    (retcode, out, err) = self.run_slider_command(" ".join([self.slider_exec, "version"]), self.slider_user)
    if retcode != 0:
      raise Exception("Could not execute version check using " + self.slider_exec)

    if not "Compiled against Hadoop" in out:
      raise Exception("Could not execute version check using " + self.slider_exec)

    self.agent_conf = self.findfile("agent.ini", self.slider_root)
    self.agent_tarball = self.findfile("slider-agent*.tar.gz", self.slider_root)

    pass

  def configure_slider_client(self):
    slider_conf_file = self.findfile("slider-client.xml", self.slider_root)

    data = None
    with open(slider_conf_file, 'r') as orig:
      data = orig.readlines()

    # add/replace a set of well known configs

    name_value_to_add = {
      SliderTesterConfig.SLIDER_ZK_QUORUM:
        self.config.get(SliderTesterConfig.CLUSTER_SECTION, SliderTesterConfig.SLIDER_ZK_QUORUM),
      SliderTesterConfig.YARN_RESOURCEMANAGER_ADDRESS:
        self.config.get(SliderTesterConfig.CLUSTER_SECTION, SliderTesterConfig.YARN_RESOURCEMANAGER_ADDRESS),
      SliderTesterConfig.YARN_RESOURCEMANAGER_SCHEDULER_ADDRESS:
        self.config.get(SliderTesterConfig.CLUSTER_SECTION, SliderTesterConfig.YARN_RESOURCEMANAGER_SCHEDULER_ADDRESS),
      SliderTesterConfig.FS_DEFAULTFS:
        self.config.get(SliderTesterConfig.CLUSTER_SECTION, SliderTesterConfig.FS_DEFAULTFS),
      SliderTesterConfig.YARN_APP_CP:
        self.config.get(SliderTesterConfig.CLUSTER_SECTION, SliderTesterConfig.YARN_APP_CP),
    }
    output = []
    for line in data:
      output.append(line)
      if "<configuration>" in line:
        output.append(os.linesep)
        logger.info("Adding additional configuations ...")
        for (k, v) in name_value_to_add.items():
          output.append("  <property>" + os.linesep)
          output.append("    <name>{0}</name>".format(k) + os.linesep)
          output.append("    <value>{0}</value>".format(v) + os.linesep)
          output.append("  </property>" + os.linesep)
          output.append(os.linesep)
        pass
      pass

    with open(slider_conf_file, 'w') as modified:
      modified.writelines(output)
    pass

  def create_hdfs_folders(self):
    self.hdfs_work_root = self.config.get(SliderTesterConfig.TEST_SECTION, SliderTesterConfig.HDFS_ROOT_DIR)
    (ret, out, err) = self.run_os_command(" ".join([SliderTester.HDFS_LS, self.hdfs_work_root]), self.hdfs_user)
    if ret == 0:
      self.run_os_command(" ".join([SliderTester.HDFS_RMR, self.hdfs_work_root]), self.hdfs_user)

    self.run_os_command(" ".join([SliderTester.HDFS_MKDIR, self.hdfs_work_root]), self.hdfs_user)
    self.run_os_command(" ".join([SliderTester.HDFS_CHOWN, self.slider_user, self.hdfs_work_root]), self.hdfs_user)

    self.hdfs_user_root = os.path.join(
      self.config.get(SliderTesterConfig.TEST_SECTION, SliderTesterConfig.HDFS_USER_DIR), self.slider_user)

    (ret, out, err) = self.run_os_command(" ".join([SliderTester.HDFS_LS, self.hdfs_user_root]), self.hdfs_user)
    if ret != 0:
      self.run_os_command(" ".join([SliderTester.HDFS_MKDIR, self.hdfs_user_root]), self.hdfs_user)
      self.run_os_command(" ".join([SliderTester.HDFS_CHOWN, self.slider_user, self.hdfs_user_root]), self.hdfs_user)

    self.hdfs_agent_root = os.path.join(self.hdfs_work_root, "agent")
    self.hdfs_agent_conf_root = os.path.join(self.hdfs_agent_root, "conf")
    self.run_os_command(" ".join([SliderTester.HDFS_MKDIR, self.hdfs_agent_root]), self.slider_user)
    self.run_os_command(" ".join([SliderTester.HDFS_MKDIR, self.hdfs_agent_conf_root]), self.slider_user)

    self.run_os_command("hdfs dfs -copyFromLocal " + self.agent_tarball + " " + self.hdfs_agent_root, self.slider_user)
    self.run_os_command("hdfs dfs -copyFromLocal " + self.agent_conf + " " + self.hdfs_agent_conf_root,
                        self.slider_user)

    self.run_os_command("hdfs dfs -copyFromLocal " + self.app_package + " " + self.hdfs_work_root, self.slider_user)

    self.agent_tarball_hdfs = os.path.join(self.hdfs_agent_root, os.path.basename(self.agent_tarball))
    self.agent_conf_hdfs = os.path.join(self.hdfs_agent_conf_root, os.path.basename(self.agent_conf))
    self.app_package_hdfs = os.path.join(self.hdfs_work_root, os.path.basename(self.app_package))
    self.cluster_name = self.config.get(SliderTesterConfig.TEST_SECTION, SliderTesterConfig.CLUSTER_NAME)

    self.cluster_location = os.path.join(self.hdfs_user_root, ".slider/cluster", self.cluster_name)
    self.run_os_command(" ".join([SliderTester.HDFS_RMR, self.cluster_location]), self.hdfs_user)

  pass

  def create_cluster(self):
    self.fsdefault = self.config.get(SliderTesterConfig.CLUSTER_SECTION, SliderTesterConfig.FS_DEFAULTFS)

    cmd = SliderTester.SLIDER_CREATE_CMD_FORMAT.format(self.slider_exec,
                                                       self.cluster_name, self.fsdefault, self.agent_tarball_hdfs,
                                                       self.agent_conf_hdfs, self.app_conf, self.app_resources,
                                                       self.app_package_hdfs)

    (retcode, out, err) = self.run_slider_command(cmd, self.slider_user)
    if retcode != 0:
      raise Exception("Could not create cluster. Out: " + out + " Err: " + err)
    pass

  def verify_cluster(self):
    (retcode, out, err) = self.run_os_command(" ".join([SliderTester.HDFS_LS, self.cluster_location]), self.slider_user)
    if retcode != 0:
      raise Exception("Could not verify cluster. Out: " + out + " Err: " + err)
    pass

  def clean_up(self):
    (retcode, out, err) = self.run_slider_command(" ".join([self.slider_exec, "stop", self.cluster_name]),
                                                  self.slider_user)
    if retcode != 0:
      raise Exception("Could not clean cluster. Out: " + out + " Err: " + err)
    pass


  def findfile(self, filepattern, path):
    matches = []
    for root, dirnames, filenames in os.walk(path):
      for filename in fnmatch.filter(filenames, filepattern):
        matches.append(os.path.join(root, filename))
    if len(matches) > 0:
      return matches[0]
    else:
      return None
    pass


  def do_test_setup(self):
    self.cluster_type = self.config.get(SliderTesterConfig.TEST_SECTION, SliderTesterConfig.CLUSTER_TYPE)
    if self.cluster_type.lower() == "habse":
      self.hbase_setup()
      return

    if self.cluster_type.lower() == "storm":
      self.storm_setup()
      return

    if self.cluster_type.lower() == "storm":
      self.storm_setup()
      return
    pass


  def hbase_setup(self):
    (ret, out, err) = self.run_os_command(" ".join([SliderTester.HDFS_LS, "/app"]), self.hdfs_user)
    if ret != 0:
      self.run_os_command(" ".join([SliderTester.HDFS_MKDIR, "/app"]), self.hdfs_user)
    pass

    (ret, out, err) = self.run_os_command(" ".join([SliderTester.HDFS_LS, "/app/habse"]), self.hdfs_user)
    if ret == 0:
      self.run_os_command(" ".join([SliderTester.HDFS_RMR, "/app/hbase"]), self.hdfs_user)

    self.run_os_command(" ".join([SliderTester.HDFS_MKDIR, "/app/hbase"]), self.hdfs_user)
    self.run_os_command(" ".join([SliderTester.HDFS_CHOWN, self.slider_user, "/app/hbase"]), self.hdfs_user)
    pass

  pass


def resolve_config(configFile):
  try:
    config = SliderTesterConfig.config
    if os.path.exists(configFile):
      config.read(configFile)
      SliderTesterConfig.setConfig(config)
    else:
      raise Exception("No config found, use default")

  except Exception, err:
    logger.warn(err)
  return config


def main():
  parser = OptionParser()
  parser.add_option("-c", "--config", dest="config", help="SliderTester config file location", default=None)
  parser.add_option("-o", "--out", dest="outputfile", default="/tmp/slider-test.log", help="log file to store results.",
                    metavar="FILE")
  (options, args) = parser.parse_args()

  if os.path.isfile(options.outputfile):
    os.remove(options.outputfile)

  print "Logging at " + options.outputfile

  global logger
  logger = logging.getLogger('HostCleanup')
  handler = logging.FileHandler(options.outputfile)
  formatter = logging.Formatter('%(asctime)s %(levelname)s %(message)s')
  handler.setFormatter(formatter)
  logger.addHandler(handler)
  logging.basicConfig(level=logging.DEBUG)

  if options.config:
    resolve_config(options.config)

  s = SliderTester(SliderTesterConfig().getConfig())

  s.deploy_and_validate_slider()

  s.configure_slider_client()

  s.create_hdfs_folders()

  s.do_test_setup()

  s.create_cluster()

  time.sleep(120)
  s.verify_cluster()

  s.clean_up();

  # Copy slider and app package to TEST_ROOT/packages
  # Expand app package at TEST_ROOT/app
  # Expand slider package at TEST_ROOT/slider
  ## set jdk path
  ## check slider version

  # Create HDFS folders and ensure permission
  # Populate HDFS folders
  # Finalize resources and appConf json files
  # Call create
  # Validate existence of the app
  # Call start


if __name__ == "__main__":
  main()