#!/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 os, sys, re, tempfile, shutil, pickle, getpass, subprocess, time, glob, ConfigParser
from xml.dom import minidom

try:
    from optparse import Option, OptionParser 
    from gppylib.gpparseopts import OptParser, OptChecker
    from gppylib.gplog import get_default_logger, setup_tool_logging
    from gppylib.commands.unix import getLocalHostname, getUserName, SYSTEM
    from gppylib.commands.base import WorkerPool, Command, REMOTE
    from gppylib.gpcheckutil import HostType, hosttype_str
    from hawqpylib.hawqlib import remote_ssh_output
    from pygresql.pgdb import DatabaseError
    from pygresql import pg
    import stat
    from gppylib.gpsqlUtil import GpsqlUtil
    import yaml
    import xml
    import time
    import calendar

except ImportError, e:    
    sys.exit('Cannot import modules.  Please check that you have sourced greenplum_path.sh.  Detail: ' + str(e))


class GpCheckError(Exception):
    pass


class GpCheckInfo:
    def __init__(self):
        self.is_root = (os.geteuid() == 0)
        self.host_type = HostType.GPCHECK_HOSTTYPE_UNDEFINED
        self.appliance_version = None

        # record gpcheck_hostdump data for each host
        self.hosts = dict() # hostname => GpCheckHost obj

        # record HAWQ configuration
        self.hawq_gucs = dict() # guc name => (master_value, segment_value)
        self.hawq_segment_configuration = None
        self.hawq_collected_ok = False # if successfully collect HAWQ gucs

        self.collection_start_time = 0 # used in NTPD testing
        self.collection_end_time = 0 # used in NTPD testing


class GpCheckHost:
    def __init__(self, name, is_namenode=False):
        self.hostname = name
        self.datafile = None # pickle file on each host
        self.data = None # `gpcheck_hostdump` collected data for each host
        self.is_namenode = is_namenode

    def __str__(self):
        s = "%s datafile(%s)" % (self.hostname, self.datafile)
        if self.is_namenode:
            s += " namenode"
        return s


class GpCheckConfig:
    def __init__(self):
        self.parser = ConfigParser.RawConfigParser()
        self.gpcheck_config_version = 0

        self.mount_points = set()
        self.sysctl_expected = dict()

        self.limits_expected = { # default value for limits.conf
                ("soft", "nofile"): 2900000,
                ("hard", "nofile"): 2900000,
                ("soft", "nproc") : 131072,
                ("hard", "nproc") : 131072 }

        self.diskusage_mounts = []
        self.diskusage_usagemax = 90 # max disk usage percentage
        self.disk_space_total = 1000.0

        self.hdfs_expected = { # default value for HDFS configuration
                "dfs.mem.namenode.heap": 8192,
                "dfs.mem.datanode.heap": 8192 }
        self.hdfs_non_expected = {}
        self.hdfs_ha_expected = {}
        self.hdfs_kerberos_expected = {}
        self.hdfs_ha_kerberos_expected = {}

        self.yarn_expected = {}
        self.yarn_non_expected = {}
        self.yarn_ha_expected = {}
        self.yarn_kerberos_expected = {}
        self.yarn_ha_kerberos_expected = {}

        self.hawq_expected = {}
        self.hawq_kerberos_expected = {}
        self.hawq_yarn_expected = {}

        self.default_hash_table_bucket_number = 8
        self.hawq_rm_nvseg_perquery_limit = 512
        self.hawq_rm_nvseg_perquery_perseg_limit = 8
        self.magma_shm_limit_per_block = 1

        self.disk_write_min_bandwidth = 200.0
        self.disk_read_min_bandwidth = 700.0
        self.stream_min_bandwidth = 10000.0
        self.network_min_bandwidth = 1000.0


    def readConfigFile(self, config_file):
        parsed_list = self.parser.read(config_file)
        if len(parsed_list) != 1:
            raise GpCheckError("cannot open file!")

        if not self.parser.has_section("linux.sysctl"):
            raise GpCheckError("require section 'linux.sysctl'")

        section = "global"
        if self.parser.has_option(section, "configfile_version"):
            self.gpcheck_config_version = self.parser.getint(section, "configfile_version")

        section = "linux.mount"
        if self.parser.has_option(section, "mount.points"):
            for p in self.parser.get(section, "mount.points").split(","):
                self.mount_points.add(p.strip())

        section = 'linux.sysctl'
        for opt in self.parser.options(section):
            if re.match('sysctl\.', opt):
                fields = opt.split('sysctl.')
                if len(fields) != 2:
                    raise GpCheckError("Bad config line entry '%s'" % opt)
                self.sysctl_expected[fields[1]] = self.parser.get(section, opt)

        section = "linux.limits"
        for opt in self.parser.options(section):
            key = tuple(opt.split("."))
            self.limits_expected[key] = self.parser.getint(section, opt)

        section = "linux.diskusage"
        if self.parser.has_option(section, "diskusage.monitor.mounts"):
            self.diskusage_mounts = [m.strip() for m in self.parser.get(section, "diskusage.monitor.mounts").split(",")]
        if self.parser.has_option(section, "diskusage.monitor.usagemax"):
            self.diskusage_usagemax = self.parser.get(section, "diskusage.monitor.usagemax")
            try:
                if self.diskusage_usagemax[-1] == "%":
                    self.diskusage_usagemax = int(self.diskusage_usagemax[:-1])
                else:
                    self.diskusage_usagemax = int(self.diskusage_usagemax)

            except Exception, e:
                raise GpCheckError("Bad config entry value '%s' for 'diskusage.monitor.usagemax': %s" %
                        (self.diskusage_usagemax, e))
        if self.parser.has_option(section, "disk.space.total.GB"):
            try:
                self.disk_space_total = float(self.parser.get(section, "disk.space.total.GB"))
            except expression as e:
                raise GpCheckError("Fail to read section 'linux.diskusage': %s" % (str(e)))

        if not self.parser.has_section('hdfs.base'):
            if not self.parser.has_section("hdfs"):
                raise GpCheckError("require section 'hdfs'")

            section = 'hdfs'
            for opt in self.parser.options(section):
                self.hdfs_expected[opt] = self.parser.get(section, opt)
            try:
                self.hdfs_expected["dfs.mem.namenode.heap"] = int(self.hdfs_expected["dfs.mem.namenode.heap"])
                self.hdfs_expected["dfs.mem.datanode.heap"] = int(self.hdfs_expected["dfs.mem.datanode.heap"])
            except ValueError, e:
                raise GpCheckError("'dfs.mem.namenode.heap' or 'dfs.mem.namenode.heap' should be a number: %s" % e)
        else:
            section = 'hdfs.base'
            for opt in self.parser.options(section):
                self.hdfs_expected[opt] = self.parser.get(section, opt)
            try:
                self.hdfs_expected["dfs.mem.namenode.heap"] = int(self.hdfs_expected["dfs.mem.namenode.heap"])
                self.hdfs_expected["dfs.mem.datanode.heap"] = int(self.hdfs_expected["dfs.mem.datanode.heap"])
            except ValueError, e:
                raise GpCheckError("'dfs.mem.namenode.heap' or 'dfs.mem.namenode.heap' should be a number: %s" % e)

            section = 'hdfs.non'
            for opt in self.parser.options(section):
                self.hdfs_non_expected[opt] = self.parser.get(section, opt)

            section = 'hdfs.ha'
            for opt in self.parser.options(section):
                self.hdfs_ha_expected[opt] = self.parser.get(section, opt)

            section = 'hdfs.kerberos'
            for opt in self.parser.options(section):
                self.hdfs_kerberos_expected[opt] = self.parser.get(section, opt)

            section = 'hdfs.ha.kerberos'
            for opt in self.parser.options(section):
                self.hdfs_ha_kerberos_expected[opt] = self.parser.get(section, opt)

            section = 'yarn.base'
            for opt in self.parser.options(section):
                self.yarn_expected[opt] = self.parser.get(section, opt)

            section = 'yarn.non'
            for opt in self.parser.options(section):
                self.yarn_non_expected[opt] = self.parser.get(section, opt)

            section = 'yarn.ha'
            for opt in self.parser.options(section):
                self.yarn_ha_expected[opt] = self.parser.get(section, opt)

            section = 'yarn.kerberos'
            for opt in self.parser.options(section):
                self.yarn_kerberos_expected[opt] = self.parser.get(section, opt)

            section = 'yarn.ha.kerberos'
            for opt in self.parser.options(section):
                self.yarn_ha_kerberos_expected[opt] = self.parser.get(section, opt)

            section = 'hawq.base'
            for opt in self.parser.options(section):
                self.hawq_expected[opt] = self.parser.get(section, opt)

            section = 'hawq.kerberos'
            for opt in self.parser.options(section):
                self.hawq_kerberos_expected[opt] = self.parser.get(section, opt)

            section = 'hawq.yarn'
            for opt in self.parser.options(section):
                self.hawq_yarn_expected[opt] = self.parser.get(section, opt)

        section = "hawq.guc"
        if self.parser.has_option(section, "default.hash.table.bucket.number"):
            try:
                self.default_hash_table_bucket_number = int(self.parser.get(section, "default.hash.table.bucket.number"))
            except expression as e:
                raise GpCheckError("Fail to read section 'hawq.guc': %s" % (str(e)))
        if self.parser.has_option(section, "hawq.rm.nvseg.perquery.limit"):
            try:
                self.hawq_rm_nvseg_perquery_limit = int(self.parser.get(section, "hawq.rm.nvseg.perquery.limit"))
            except expression as e:
                raise GpCheckError("Fail to read section 'hawq.guc': %s" % (str(e)))
        if self.parser.has_option(section, "hawq.rm.nvseg.perquery.perseg.limit"):
            try:
                self.hawq_rm_nvseg_perquery_perseg_limit = int(self.parser.get(section, "hawq.rm.nvseg.perquery.perseg.limit"))
            except expression as e:
                raise GpCheckError("Fail to read section 'hawq.guc': %s" % (str(e)))
        if self.parser.has_option(section, "magma.shm.limit.per.block"):
            try:
                self.magma_shm_limit_per_block = int(self.parser.get(section, "magma.shm.limit.per.block"))
            except expression as e:
                raise GpCheckError("Fail to read section 'hawq.guc': %s" % (str(e)))

        section = "hardware.performance"
        if self.parser.has_option(section, "disk.write.min.bandwidth.MBs"):
            try:
                self.disk_write_min_bandwidth = float(self.parser.get(section, "disk.write.min.bandwidth.MBs"))
            except expression as e:
                raise GpCheckError("Fail to read section 'hardware.performance': %s" % (str(e)))
        if self.parser.has_option(section, "disk.read.min.bandwidth.MBs"):
            try:
                self.disk_read_min_bandwidth = float(self.parser.get(section, "disk.read.min.bandwidth.MBs"))
            except expression as e:
                raise GpCheckError("Fail to read section 'hardware.performance': %s" % (str(e)))
        if self.parser.has_option(section, "stream.min.bandwidth.MBs"):
            try:
                self.stream_min_bandwidth = float(self.parser.get(section, "stream.min.bandwidth.MBs"))
            except expression as e:
                raise GpCheckError("Fail to read section 'hardware.performance': %s" % (str(e)))
        if self.parser.has_option(section, "network.min.bandwidth.MBs"):
            try:
                self.network_min_bandwidth = float(self.parser.get(section, "network.min.bandwidth.MBs"))
            except expression as e:
                raise GpCheckError("Fail to read section 'hardware.performance': %s" % (str(e)))

###### Global Variables #############
logger = get_default_logger()
EXECNAME = os.path.split(__file__)[-1]
setup_tool_logging(EXECNAME,getLocalHostname(),getUserName())

options = None
GPHOME = None
GPCHECK_CONFIG_FILE = None
HADOOP_HOME = None

gpcheck_info = GpCheckInfo()
gpcheck_config = GpCheckConfig()
pool = WorkerPool()
tmpdir = None
found_errors = 0

HAWQ_GUC_MEMORY = "hawq_re_memory_overcommit_max"


def checkPlatform():
    host_type_map = { "linux": HostType.GPCHECK_HOSTTYPE_GENERIC_LINUX,
                      "sunos": HostType.GPCHECK_HOSTTYPE_GENERIC_SOLARIS }
    try:
        gpcheck_info.host_type = host_type_map[SYSTEM.getName()]
        logger.info("Detected platform: %s" % hosttype_str(gpcheck_info.host_type))

    except KeyError:
        raise GpCheckError("No tests exists for this platform in gpcheck")


def parse_host_list_file(host_file):
    host_list = list()
    with open(host_file) as f:
        hosts = f.readlines()
    for host in hosts:
        host = host.split("#",1)[0].strip()
        if host:
            host_list.append(host)
    return host_list

def parseargs():
    global options, GPHOME, HADOOP_HOME, GPCHECK_CONFIG_FILE

    parser = OptParser(option_class=OptChecker, version='%prog version $Revision: #1 $')
    parser.remove_option('-h')
    parser.add_option('-?', '--help', action='help')
    parser.add_option('--verbose', action='store_true')
    parser.add_option('--stdout', action='store_true')
    parser.add_option('--zipout', action='store_true')
    parser.add_option('--zipin', type='string')
    parser.add_option('--gphome', type='string')
    # for HDFS xml and memory check
    parser.add_option('--hadoop', '--hadoop-home', type='string')
    parser.add_option('--hdfs', action='store_true')
    parser.add_option('--hdfs-ha', dest="hdfs_ha", action='store_true')
    parser.add_option('--yarn', action='store_true')
    parser.add_option('--yarn-ha', dest="yarn_ha", action='store_true')
    parser.add_option('--kerberos', action='store_true')

    parser.add_option('-c', '--config', type='string') # optional: gpcheck config file path
    parser.add_option('-f', '--file',  type='string')  # host file, for testing a list of hosts
    parser.add_option('-h', '--host',  type='string')  # test a single host

    # Adding new options: postSetup postUpgrade weeklyExam
    parser.add_option('--postSetup', action='store_true')
    parser.add_option('--postUpgrade', action='store_true')
    parser.add_option('--weeklyExamine', action='store_true')
    parser.add_option('-t', '--table', type='string') # optional: table to analyze

    (options, args) = parser.parse_args()
    if len(args) > 0:
        if args[0] == 'help':
            parser.print_help(sys.stderr)
            sys.exit(0)

    # GPHOME must be found
    GPHOME = options.gphome if options.gphome else os.environ.get("GPHOME")
    if not GPHOME:
        raise GpCheckError("GPHOME not set, must be specified in --gphome")
    GPCHECK_CONFIG_FILE = options.config if options.config else "%s/etc/gpcheck.cnf" % GPHOME
    logger.info("Checks uses config file: %s", GPCHECK_CONFIG_FILE)

    HADOOP_HOME = options.hadoop if options.hadoop else os.environ.get("HADOOP_HOME")

    if not HADOOP_HOME:
        checkFailed(None, "utility will SKIP HDFS configuration check because HADOOP_HOME is not specified in environment variable or --hadoop")

    if options.yarn and not HADOOP_HOME:
        options.yarn = False
        checkFailed(None, "utility will SKIP YARN configuration check because HADOOP_HOME is not specified in environment variable or --hadoop")

    # params check
    if not options.file and not options.host and not options.zipin:
        raise GpCheckError(" --file or --host or --zipin must be specified")

    if options.file and options.host:
        raise GpCheckError(" You can specify either --file or --host, but not both")

    if options.stdout and options.zipout:
        raise GpCheckError(" You can specify either --stdout or --zipout, but not both")


def readConfigFile():
    try:
        gpcheck_config.readConfigFile(GPCHECK_CONFIG_FILE)

    except Exception, e:
        raise GpCheckError("Field to read gpcheck config file '%s':\n%s" % (GPCHECK_CONFIG_FILE, e))


def checkFailed(host, msg):
    global found_errors
    found_errors += 1
    if host:
        logger.error("host(%s): %s", host, msg)
    else:
        logger.error(msg)


def getHDFSNamenodeHost():
    core_site_file = os.path.join(HADOOP_HOME, "etc/hadoop/core-site.xml")
    hdfs_site_file = os.path.join(HADOOP_HOME, "etc/hadoop/hdfs-site.xml")
    logger.info("try to detect namenode from %s" % core_site_file)

    # for processing property xml
    getPropName =  lambda node: node.getElementsByTagName('name')[0].childNodes[0].data
    getPropValue = lambda node: node.getElementsByTagName('value')[0].childNodes[0].data

    # read namenode address from core-site.xml
    with open(core_site_file) as f:
        xmldoc = minidom.parse(f)
    namenode_addr = ''
    for node in xmldoc.getElementsByTagName('property'):
        if getPropName(node) == 'fs.default.name' or getPropName(node) == 'fs.defaultFS':
            fsurl = getPropValue(node).strip()
            namenode_list_alias = re.search(r"//([^:/]*)", fsurl).group(1)
            if_ha_disabled = re.search(".*:[0-9]+$", fsurl)
            if if_ha_disabled:
                namenode_addr = namenode_list_alias
            else:
                namenode_addr = ''
            break

    # run hostname command on remote to get actual hostname
    if namenode_addr == '':
        ha_namenode_list = ''
        default_namenode_alias = ''
        with open(hdfs_site_file) as f:
            xmldoc = minidom.parse(f)
        for node in xmldoc.getElementsByTagName('property'):
            if re.search('dfs.ha.namenodes.*', getPropName(node).strip()):
                ha_namenode_list = getPropValue(node).strip()
                default_namenode_alias = ha_namenode_list.split(',')[0].strip()
                break

        if ha_namenode_list == '':
            logger.error("cannot detect namenode from %s" % core_site_file)
            raise GpCheckError("cannot detect namenode from %s" % core_site_file)
            #sys.exit(1)
        else:
            with open(hdfs_site_file) as f:
                xmldoc = minidom.parse(f)
            for node in xmldoc.getElementsByTagName('property'):
                namenode_rpc_address = "dfs.namenode.rpc-address.%s.%s" % (namenode_list_alias,
                                                                           default_namenode_alias)
                if getPropName(node) == namenode_rpc_address:
                    default_namenode_rpc_address = getPropValue(node).strip()
                    namenode_addr = default_namenode_rpc_address.split(':')[0].strip()

    if namenode_addr == '':
        raise GpCheckError("cannot detect namenode from %s" % core_site_file)
    else:
        cmd = Command(namenode_addr, "hostname", REMOTE, namenode_addr)
        pool.addCommand(cmd)
        pool.join()
        items = pool.getCompletedItems()
        for i in items:
            if i.results.rc or i.results.halt or not i.results.completed:
                raise Exception("error running 'hostname' command: %s" % i.results.stderr.strip())
            namenode_host = i.results.stdout.strip()
        logger.info("detect namenode hostname to be %s" % namenode_host)
    return namenode_host

def createHostList():
    if options.verbose:
        logger.info("trying to deduplicate hosts...")

    hostlist = []
    # read the host file if present
    if options.file:
        try:
            with open(options.file, "r") as f:
                hostlist = [line.strip() for line in f.readlines() if line.strip()]

        except IOError, e:
            raise GpCheckError("error reading host file '%s': %s" % (options.file, str(e)))
    else:
        hostlist.append(options.host)

    # get actual hostname and deduplicate
    try:
        for hostname in hostlist:
            cmd = Command(hostname, "hostname", REMOTE, hostname)
            pool.addCommand(cmd)

        pool.join()
        items = pool.getCompletedItems()
        for i in items:
            if i.results.rc or i.results.halt or not i.results.completed:
                raise Exception("error running 'hostname' on host '%s': %s" % (i.remoteHost, i.results.stderr.strip()))

            actualHostname = i.results.stdout.strip()
            if actualHostname not in gpcheck_info.hosts:
                gpcheck_info.hosts[actualHostname] = GpCheckHost(actualHostname)

    except Exception, e:
        raise GpCheckError("failed to collect 'hostname' on servers: %s" % str(e))

    if options.verbose:
        logger.info("trying to deduplicate hosts [success]")

    if HADOOP_HOME:
        try:
            namenode_host = getHDFSNamenodeHost()
            if namenode_host in hostlist:
                gpcheck_info.hosts[namenode_host] = GpCheckHost(namenode_host, is_namenode=True)
            else:
                logger.warning("utility will skip HDFS namenode check since it's not in current host list.")

        except Exception, e:
            checkFailed(None, "utility will SKIP HDFS namenode check: %s" % str(e))


def runCollections():
    logger.info("trying to collect server configuration...")

    # run gpcheck_hostdump on each server
    runCollectionOnServers()
    # copy hostdump file to master
    copyFilesLocally()
    # delete hostdump file on remote servers
    deleteRemoteFiles()

    logger.info("trying to collect server configuration [success]")


def runCollectionOnServers():
    gpcheck_info.collection_start_time = time.time()

    def getDumpCommand():
        if gpcheck_info.host_type == HostType.GPCHECK_HOSTTYPE_GENERIC_LINUX:
            host_type_cl = "--linux"
        elif gpcheck_info.host_type == HostType.GPCHECK_HOSTTYPE_GENERIC_SOLARIS:
            host_type_cl = "--solaris"
        else:
            raise GpCheckError("unsupported host type")

        cmd = "%s/sbin/gpcheck_hostdump --hawq %s" % (GPHOME, host_type_cl)
        cmd += " --sysctl %s" % ",".join(gpcheck_config.sysctl_expected.keys())
        if HADOOP_HOME:
            cmd += " --hadoop %s" % HADOOP_HOME
        if options.yarn or options.yarn_ha:
            cmd += " --yarn"
        return cmd

    try:
        cmdStr = getDumpCommand()
        for host in gpcheck_info.hosts:
            if options.verbose:
                logger.info("collect data on host: %s" % host)

            cmd = Command(host, cmdStr, REMOTE, host)
            pool.addCommand(cmd)

        pool.join()
        items = pool.getCompletedItems()
        for i in items:
            if i.results.rc or i.results.halt or not i.results.completed:
                raise Exception("error running gpcheck_hostdump on '%s': %s" % (i.remoteHost, i.results.stderr.strip()))

            gpcheck_info.hosts[i.remoteHost].datafile = i.results.stdout.strip()
            
    except Exception, e:
        raise GpCheckError("Failed to collect data from servers:\n%s" % e)

    gpcheck_info.collection_end_time = time.time()


def copyFilesLocally():
    if options.verbose:
        logger.info("copy hostdump files from remote servers to master")

    try:
        for host in gpcheck_info.hosts:
            cmdStr = "scp %s:%s %s/%s.data" % (host, gpcheck_info.hosts[host].datafile, tmpdir, host)
            if options.verbose:
                logger.info(cmdStr)
            cmd = Command(host, cmdStr)
            pool.addCommand(cmd)

        pool.join()
        items = pool.getCompletedItems()
        for i in items:
            if i.results.rc or i.results.halt or not i.results.completed:
                raise Exception("error running command %s: %s" % (i.cmdStr, i.results.stderr.strip()))

    except Exception, e:
        raise GpCheckError("Failed to scp remote hostdump file to master:\n%s" % e)


def deleteRemoteFiles():
    if options.verbose:
        logger.info("delete hostdump files on remote servers")

    try:
        for host in gpcheck_info.hosts:
            cmdStr = "rm -f %s" % gpcheck_info.hosts[host].datafile
            if options.verbose:
                logger.info(cmdStr)
            cmd = Command(host, cmdStr, REMOTE, host)
            pool.addCommand(cmd)

        pool.join()
        items = pool.getCompletedItems()
        for i in items:
            if i.results.rc or i.results.halt or not i.results.completed:
                raise Exception("error running command %s: %s" % (i.cmdStr, i.results.stderr.strip()))

    except Exception, e:
        raise GpCheckError("Failed to delete remote hostdump file:\n%s" % e)


def readDataFiles():
    for host in gpcheck_info.hosts:
        fname = "%s/%s.data" % (tmpdir, host)
        try:
            with open(fname, "rb") as f:
                gpcheck_info.hosts[host].data = pickle.load(f)

        except Exception, e:
            raise GpCheckError("Failed to load pickle file '%s': %s" % (fname, e))


def readHAWQConfiguration():
    '''
    This function is used to check hawq_re_memory_overcommit_max parameter.

    Sets the maximum quota of memory overcommit (in MB) per physical segment for resource enforcement. 

    This parameter sets the memory quota that can be overcommited beyond the memory quota dynamically assigned by the resource manager.
    '''
    if options.verbose:
        logger.info("trying to collect HAWQ configuration...")

    try:
        _sqlUtil = GpsqlUtil()
        gpcheck_info.hawq_segment_configuration = _sqlUtil.performQuery("select * from gp_segment_configuration;")
    except expression as identifier:
        logger.error("utility cannot perform HAWQ CPU and Memory check because failed to connect to HAWQ")
        logger.error(str(e))

    # read Memory GUC using hawqconfig
    command = "hawqconfig -s %s" % HAWQ_GUC_MEMORY
    p = subprocess.Popen(command, shell = True,
            stdout = subprocess.PIPE, stderr = subprocess.PIPE)
    result = p.communicate()
    match_master = re.search(r'Value		: (\d+)', result[0])

    if match_master:
        gpcheck_info.hawq_gucs[HAWQ_GUC_MEMORY] = (int(match_master.group(1)))
    else:
        checkFailed(None, "utility cannot perform HAWQ Memory check because failed to get GUC value using '%s'" % command)
        return

    gpcheck_info.hawq_collected_ok = True
    if options.verbose:
        logger.info("trying to collect HAWQ configuration [success]")


def testConnectEmc(host):
    if not host.is_a_master:
        return
    
    expected = "Running"

    if host.data.connectemc.output != expected:
        checkFailed(host.hostname, "Connect EMC is not running on master (try /etc/init.d/connectemc status)")


def testSolarisEtcSystem(host):
    requiredValues = {  'rlim_fd_cur' : '65536',
                        'zfs:zfs_arc_max' : '0x600000000',
                        'pcplusmp:apic_panic_on_nmi' : '1',
                        'nopanicdebug' : '1' }

    results = dict()

    for k in requiredValues.keys():
        results[k] = 0
        
    for key in host.data.etc_system.parameters.keys():

        if key not in requiredValues:
            continue

        foundValue = host.data.etc_system.parameters[key]
        if foundValue == requiredValues[key]:
            results[key] = 1

    for k in results.keys():

        if results[k]:
            continue

        checkFailed(host.hostname, "/etc/system is missing expected line 'set %s=%s'" % (k, requiredValues[k]))


def testSolarisEtcProject(host):

    requiredValues = {  'default:3::::project.max-sem-ids=(priv,1024,deny);process.max-file-descriptor=(priv,252144,deny)' : 0 }

    unexpectedValues = set(['default:3::::'])

    for line in host.data.etc_project.lines:
        if line in unexpectedValues:
            checkFailed(host.hostname, "unexpected line in /etc/project: '%s'" % line)
            continue

        if line in requiredValues:
            requiredValues[line] = 1

    for line in requiredValues.keys():
        if requiredValues[line]:
            continue

        checkFailed(host.hostname, "/etc/project is missing expected line '%s'" % line)


def testSolarisEtcUserAttr(host):

    requiredValues = { 'gpadmin::::defaultpriv=basic,dtrace_user,dtrace_proc' : 0 }

    for line in host.data.etc_user_attr.lines:
        if line in requiredValues:
            requiredValues[line] = 1

    for line in requiredValues.keys():
        if requiredValues[line]:
            continue

        checkFailed(host.hostname, "/etc/user_attr is missing expected line '%s'" % line)


def testHAWQGUC(host):
    if not gpcheck_info.hawq_collected_ok:
        return

    if options.verbose:
        logger.info("-- test HAWQ CPU/Memory Guc Settings")

    c = gpcheck_info.hawq_segment_configuration
    master_hostname = filter(lambda x: x['role'] == 'm', c)[0]['hostname']

    if host.hostname not in map(lambda x: x['hostname'], c):
        logger.warning("host '%s' is not in HAWQ array" % host.hostname)
        return

    actual_total_memory   = host.data.machine.memory_in_MB

    guc_vmemsize_master = gpcheck_info.hawq_gucs[HAWQ_GUC_MEMORY]
    # segment count on this host
    num_segments = len(filter(lambda x: x['hostname'] == host.hostname, c))

    if host.hostname == master_hostname:
        if num_segments > 1:
            checkFailed(host.hostname, "HAWQ master host has segments configured")

        if actual_total_memory < guc_vmemsize_master:
            checkFailed(host.hostname, "HAWQ master host memory size '%s' is less than the '%s' size '%s'" % (
                actual_total_memory, HAWQ_GUC_MEMORY, guc_vmemsize_master))
            return

        # check HAWQ master's memory size
        expected_vmemory_size = 8192
        if guc_vmemsize_master != expected_vmemory_size:
            checkFailed(host.hostname, "HAWQ master's %s GUC value is %s, expected %s" % (
                HAWQ_GUC_MEMORY, guc_vmemsize_master, expected_vmemory_size))

    else:
        datanode_mem = gpcheck_config.hdfs_expected["dfs.mem.datanode.heap"]

        # check HAWQ memory size
        if actual_total_memory < datanode_mem:
            checkFailed(host.hostname, "HAWQ segment's host memory size '%s' is less than the expected data node memory size '%s'" % (
                actual_total_memory, datanode_mem))
            logger.warning("please change the expected data node memory 'dfs.mem.datanode.heap' in gpcheck.cnf file")
            logger.warning("SKIP '%s' check" %(HAWQ_GUC_MEMORY))
            return
        expect_vmemsize_per_segment = 8192 
        if guc_vmemsize_master != expect_vmemsize_per_segment:
            checkFailed(host.hostname, "HAWQ segment's %s GUC value on this host is %s, expected %s" % (
                HAWQ_GUC_MEMORY, guc_vmemsize_master, expect_vmemsize_per_segment))


def testDiskCapacity(host):
    if options.verbose:
        logger.info("-- test Disk Capacity")

    for line in host.data.diskusage.lines:
        if len(gpcheck_config.diskusage_mounts) == 0 or line.mount in gpcheck_config.diskusage_mounts:
            actual_usage = int(line.used_percent[:-1])
            if actual_usage > gpcheck_config.diskusage_usagemax:
                checkFailed(host.hostname,
                           "potential disk full risk: %s mounted on %s has used %s space" % (
                               line.fs, line.mount, line.used_percent))
    return


def testHAWQconfig(host):
    hawq = host.data.hawq
    hdfs = host.data.hdfs
    if hawq is None:
        return # skip HAWQ test when hawq is None

    if hdfs is None:
        return # skip HAWQ test when hdfs is None

    if options.verbose:
        logger.info("-- test HAWQ config")

    if hawq.errormsg:
        checkFailed(host.hostname, "collect HAWQ configuration error: %s" % hawq.errormsg)
        return

    datanode_list = list()
    if HADOOP_HOME:
        datanode_list = parse_host_list_file("%s/etc/hadoop/slaves" % HADOOP_HOME)
    is_datanode = False
    if host.hostname in datanode_list:
        is_datanode = True

    expect_config = gpcheck_config.hawq_expected

    if options.kerberos:
        expect_config.update(gpcheck_config.hawq_kerberos_expected)

    if options.yarn or options.yarn_ha:
        expect_config.update(gpcheck_config.hawq_yarn_expected)

    actual_config = hawq.site_config
    hdfs_actual_config = hdfs.site_config

    for exp_key, exp_val in expect_config.items():
        if exp_key not in actual_config:
            checkFailed(host.hostname, "HAWQ configuration missing: '%s' needs to be set to '%s'" % (exp_key, exp_val))

        else:
            actual_val = actual_config[exp_key]
            et = (exp_key, exp_val, actual_val)

            if exp_key == "dfs.block.local-path-access.user":
                if exp_val not in actual_val.split(','):
                    checkFailed(host.hostname, "HDFS configuration: '%s' should include user '%s', actual value is '%s'" % et)
            elif exp_key == "dfs.namenode.handler.count":
                if int(exp_val) > int(actual_val):
                    checkFailed(host.hostname, "HDFS configuration: '%s' should be at least '%s', actual value is '%s'" % et)
            else:
                if exp_val != actual_val:
                    checkFailed(host.hostname, "HAWQ configuration: expected '%s' for '%s', actual value is '%s'" % et)

    if not options.kerberos:
        if 'hadoop.security.authentication' in actual_config:
            if actual_config['hadoop.security.authentication'] != 'simple':
                checkFailed(host.hostname, "HAWQ configuration: expected '%s' for '%s', actual value is '%s'" % ('simple', 'hadoop.security.authentication', actual_config['hadoop.security.authentication']))

        if 'hadoop.security.authentication' in hdfs_actual_config:
            if hdfs_actual_config['hadoop.security.authentication'] != 'simple':
                checkFailed(host.hostname, "HAWQ configuration: expected '%s' for '%s', actual value is '%s'" % ('simple', 'hadoop.security.authentication', hdfs_actual_config['hadoop.security.authentication']))

    if options.yarn or options.yarn_ha:
        hawq_yarn_property_exist_list = ['hawq_rm_yarn_address', 'hawq_rm_yarn_scheduler_address', 'hawq_rm_yarn_app_name']
        for item in hawq_yarn_property_exist_list:
            if item in actual_config:
                if not actual_config[item]:
                        checkFailed(host.hostname, "HAWQ configuration: yarn.resourcemanager.address is empty")
            else:
                checkFailed(host.hostname, "HAWQ configuration: yarn.resourcemanager.address not defined")

    if 'dfs.client.read.shortcircuit' not in actual_config:
        checkFailed(host.hostname, "HAWQ configuration dfs.client.read.shortcircuit not defined")

    if 'dfs.client.read.shortcircuit' not in hdfs_actual_config:
        checkFailed(host.hostname, "HAWQ configuration dfs.client.read.shortcircuit not defined")

    if 'dfs.domain.socket.path' not in actual_config:
        checkFailed(host.hostname, "HAWQ configuration dfs.domain.socket.path not defined")

    if 'dfs.domain.socket.path' not in hdfs_actual_config:
        checkFailed(host.hostname, "HDFS configuration dfs.domain.socket.path not defined")

    if is_datanode and 'dfs.domain.socket.path' in actual_config and 'dfs.domain.socket.path' in hdfs_actual_config:
        if actual_config['dfs.domain.socket.path'] != hdfs_actual_config['dfs.domain.socket.path']:
            checkFailed(host.hostname, "HAWQ configuration: dfs.domain.socket.path expect to have the same value with HDFS configuration")
        else:
            cmd = "ls -l %s" % actual_config['dfs.domain.socket.path']
            (result, output, errmsg) = remote_ssh_output(cmd, host.hostname, '')
            if result == 0:
                if output.split(' ')[0][7:9] != 'rw':
                    checkFailed(host.hostname, "HAWQ configuration dfs.domain.socket.path: %s should have R/W access for both hawq and HDFS on %s" % (actual_config['dfs.domain.socket.path'], host.hostname))
            else:
                checkFailed(host.hostname, "HAWQ configuration dfs.domain.socket.path: %s, does not exist on %s" % (actual_config['dfs.domain.socket.path'], host.hostname))

    if 'output.replace-datanode-on-failure' in actual_config and len(datanode_list) > 0:
        if len(datanode_list) < 4:
            if actual_config['output.replace-datanode-on-failure'] == 'true':
                checkFailed(host.hostname, "HAWQ configuration: output.replace-datanode-on-failure expect false, current is true")
        else:
            if actual_config['output.replace-datanode-on-failure'] == 'false':
                checkFailed(host.hostname, "HAWQ configuration: output.replace-datanode-on-failure expect true, current is false")
    else:
        checkFailed(host.hostname, "HAWQ configuration: output.replace-datanode-on-failure not defined")


def testDiskCapacity(host):
    if options.verbose:
        logger.info("-- test Disk Capacity")

    for line in host.data.diskusage.lines:
        if len(gpcheck_config.diskusage_mounts) == 0 or line.mount in gpcheck_config.diskusage_mounts:
            actual_usage = int(line.used_percent[:-1])
            if actual_usage > gpcheck_config.diskusage_usagemax:
                checkFailed(host.hostname,
                           "potential disk full risk: %s mounted on %s has used %s space" % (
                               line.fs, line.mount, line.used_percent))
    return


def testHDFSConfig(host):
    hdfs = host.data.hdfs
    if hdfs is None:
        return # skip HDFS test when hdfs is None

    if options.verbose:
        logger.info("-- test HDFS config")

    if hdfs.errormsg:
        checkFailed(host.hostname, "collect HDFS configuration error: %s" % hdfs.errormsg)
        return

    expect_config = gpcheck_config.hdfs_expected

    if not options.hdfs_ha and not options.kerberos:
        expect_config.update(gpcheck_config.hdfs_non_expected)

    if options.hdfs_ha and not options.kerberos:
        expect_config.update(gpcheck_config.hdfs_ha_expected)

    if options.kerberos and not options.hdfs_ha:
        expect_config.update(gpcheck_config.hdfs_kerberos_expected)

    if options.kerberos and options.hdfs_ha:
        expect_config.update(gpcheck_config.hdfs_ha_kerberos_expected)


    if options.yarn or options.yarn_ha:
        expect_config.update(gpcheck_config.yarn_expected)
        if not options.yarn_ha and not options.kerberos:
            expect_config.update(gpcheck_config.yarn_non_expected)

        if options.yarn_ha:
            expect_config.update(gpcheck_config.yarn_ha_expected)
        if options.kerberos:
            expect_config.update(gpcheck_config.yarn_kerberos_expected)

    actual_config = hdfs.site_config
    actual_heap_size = hdfs.namenode_heap_size if host.is_namenode else hdfs.datanode_heap_size

    if host.data.machine.memory_in_MB < actual_heap_size:
        checkFailed(host.hostname, "host memory size '%s' is less than the java max heap size '%s'" % (host.data.machine.memory_in_MB, actual_heap_size)) 

    # test hdfs_site.xml setting
    for exp_key, exp_val in expect_config.items():
        if exp_key.startswith("dfs.mem"):
            continue # these options belongs to memory tests

        if exp_key not in actual_config:
            checkFailed(host.hostname, "HDFS configuration missing: '%s' needs to be set to '%s'" % (exp_key, exp_val))

        else:
            actual_val = actual_config[exp_key]
            et = (exp_key, exp_val, actual_val)

            if exp_key == "dfs.block.local-path-access.user":
                if exp_val not in actual_val.split(','):
                    checkFailed(host.hostname, "HDFS configuration: '%s' should include user '%s', actual value is '%s'" % et)

            elif exp_key == "dfs.namenode.handler.count":
                if int(exp_val) > int(actual_val):
                    checkFailed(host.hostname, "HDFS configuration: '%s' should be at least '%s', actual value is '%s'" % et)

            else:
                if exp_val != actual_val:
                    checkFailed(host.hostname, "HDFS configuration: expected '%s' for '%s', actual value is '%s'" % et)

    # test hadoop memory setting
    expect_namenode_heap = expect_config["dfs.mem.namenode.heap"]
    expect_datanode_heap = expect_config["dfs.mem.datanode.heap"]

    if host.is_namenode and actual_heap_size < expect_namenode_heap:
        checkFailed(host.hostname, "Namenode Java heap size is only %sM, we recommends at least %sM" %
                                  (actual_heap_size, expect_namenode_heap))

    if not host.is_namenode and actual_heap_size < expect_datanode_heap:
        checkFailed(host.hostname, "Datanode Java heap size is only %sM, expect value is %sM" %
                                  (actual_heap_size, expect_datanode_heap))


    # Check if nodemanager direcotries exists
    directory_check_list = []
    datanode_list = list()
    if HADOOP_HOME:
        datanode_list = parse_host_list_file("%s/etc/hadoop/slaves" % HADOOP_HOME)
    is_datanode = False
    if host.hostname in datanode_list:
        is_datanode = True

    if options.yarn or options.yarn_ha:
        yarn_enabled = True
    else:
        yarn_enabled = False

    if yarn_enabled and is_datanode:
        if 'yarn.nodemanager.local-dirs' in actual_config: 
            directory_check_list += actual_config['yarn.nodemanager.local-dirs'].split(',')
        else:
            checkFailed(host.hostname, "YARN configuration: yarn.nodemanager.local-dirs not defined")

        if 'yarn.nodemanager.log-dirs' in actual_config: 
            directory_check_list += actual_config['yarn.nodemanager.log-dirs'].split(',')
        else:
            checkFailed(host.hostname, "YARN configuration: yarn.nodemanager.log-dirs not defined")

    for directory in directory_check_list:
        cmd = "test -e %s" % directory
        (result, output, errmsg) = remote_ssh_output(cmd, host.hostname, '')
        if result != 0:
            checkFailed(host.hostname, "YARN nodemanager directory %s does not exist" % directory)

    # Check if resource manager property exists
    if options.yarn:
        yarn_property_exist_list = ['yarn.resourcemanager.address', 'yarn.resourcemanager.scheduler.address']

    if options.yarn_ha:
        yarn_property_exist_list = ['yarn.resourcemanager.hostname.rm1', 'yarn.resourcemanager.hostname.rm2']

    if yarn_enabled:
        for item in yarn_property_exist_list:
            if item in actual_config:
                if not actual_config[item]:
                        checkFailed(host.hostname, "YARN configuration: %s is empty" % item)
            else:
                checkFailed(host.hostname, "YARN configuration: %s not defined" % item)

    # Check yarn kerberos properties
    if yarn_enabled and options.kerberos:
        yarn_kerberos_check_list = ['yarn.nodemanager.keytab', 'yarn.nodemanager.principal', \
                                    'yarn.resourcemanager.keytab', 'yarn.resourcemanager.principal']
        for item in yarn_kerberos_check_list:
            if item in actual_config:
                if not actual_config[item]:
                        checkFailed(host.hostname, "YARN configuration: %s is empty, expected non-empty" % item)
            else:
                checkFailed(host.hostname, "YARN configuration missing: %s" % item)


def testIOSchedulers(host):
    if options.verbose:
        logger.info("-- test IO scheduler")

    if host.data.ioschedulers.errormsg:
        checkFailed(host.hostname, "collect IO scheduler data error: %s" % host.data.ioschedulers.errormsg)
        return

    expectedScheduler = "deadline"
    for dev in host.data.ioschedulers.devices:
        scheduler = host.data.ioschedulers.devices[dev]
        if scheduler != expectedScheduler:
            checkFailed(host.hostname,
                       "on device (%s) IO scheduler '%s' does not match expected value '%s'" % (dev, scheduler, expectedScheduler))


# perform this test only run as root
def testBlockdev(host):
    if host.data.blockdev is None:
        return

    if options.verbose:
        logger.info("-- test block device readahead value")

    expectedReadAhead = "16384"
    for dev in host.data.blockdev.ra:
        ra = host.data.blockdev.ra[dev]
        if ra != expectedReadAhead:
            checkFailed(host.hostname,
                       "on device (%s) blockdev readahead value '%s' does not match expected value '%s'" % (dev, ra, expectedReadAhead))


def testSysctl(host):
    if options.verbose:
        logger.info("-- test sysctl value")

    if host.data.sysctl.errormsg:
        checkFailed(host.hostname, "collect sysctl params error: %s" % host.data.sysctl.errormsg)
        return

    expected_values = gpcheck_config.sysctl_expected
    real_values = host.data.sysctl.variables

    # gpcheck.conf specify a lowerbound value for these params, actual value can be larger
    params_with_lowerbound = set() # sysctl params' value must be exactly the same

    for k in expected_values:
        if k in params_with_lowerbound:
            if int(real_values[k]) < int(expected_values[k]):
                checkFailed(host.hostname,
                        "sysctl value for key '%s' has value '%s', but we expect at least '%s'" % (k, real_values[k], expected_values[k]))

        elif real_values[k] != expected_values[k]: # for other params, we expect the actual value to be the same value
            checkFailed(host.hostname,
                    "sysctl value for key '%s' has value '%s' and expects '%s'" % (k, real_values[k], expected_values[k]))


def testLimitsConf(host):
    if options.verbose:
        logger.info("-- test /etc/security/limits.conf")

    if host.data.limitsconf.errormsg:
        checkFailed(host.hostname, "collect limits.conf data error: %s" % host.data.limitsconf.errormsg)
        return

    # both dict has the form: (type, item) => value
    expect_data = gpcheck_config.limits_expected
    actual_data = dict([((e.type, e.item), e.value) for e in host.data.limitsconf.lines if e.domain in ("gpadmin", "*")])
    expect_keyset = set(expect_data.keys())
    actual_keyset = set(actual_data.keys())

    for key in expect_keyset.intersection(actual_keyset):
        expect_val = int(expect_data[key])
        actual_val = int(actual_data[key])
        if actual_val < expect_val:
            checkFailed(host.hostname,
                       "%s in  /etc/security/limits.conf has value %d lower than expected value %d" % (
                            " ".join(key), actual_val, expect_val))

    for key in expect_keyset.difference(actual_keyset):
        checkFailed(host.hostname,
                   "%s not found in /etc/security/limits.conf" % " ".join(key))


def testLinuxMounts(host):
    if options.verbose:
        logger.info("-- test mount points")

    expected_mount_points = gpcheck_config.mount_points
    actual_mount_points = set([m.dir for m in host.data.mounts.entries.values()])

    if len(expected_mount_points) == 0:
        if options.verbose:
            logger.info("-- you didn't specify any mount points to be check in %s, ignore this test" % GPCHECK_CONFIG_FILE)
        return

    if not actual_mount_points.issuperset(expected_mount_points):
        for failed_mount in expected_mount_points.difference(actual_mount_points):
            checkFailed(host.hostname, "%s is not mounted" % failed_mount) 


def testNtp(host):
    if options.verbose:
        logger.info("-- test NTP")

    if host.data.ntp.currenttime < (gpcheck_info.collection_start_time - 1):
        checkFailed(host.hostname, "potential NTPD issue.  gpcheck start time (%s) time on machine (%s)" % (time.ctime(gpcheck_info.collection_start_time), time.ctime(host.data.ntp.currenttime)))
    if host.data.ntp.currenttime > (gpcheck_info.collection_end_time + 1):
        checkFailed(host.hostname, "potential NTPD issue.  gpcheck end time (%s) time on machine (%s)" % (time.ctime(gpcheck_info.collection_start_time), time.ctime(host.data.ntp.currenttime)))
    if not host.data.ntp.running:
        checkFailed(host.hostname, "ntpd not detected on machine")


def testGenericLinuxHost(host):
    logger.info("test on host: %s" % host.hostname)
    if host.is_namenode:
        testHAWQGUC(host)
        testHAWQconfig(host)
        testHDFSConfig(host)
        testDiskCapacity(host)
        testSysctl(host)
        testLimitsConf(host)
        testLinuxMounts(host)
        testNtp(host)

    else:
        testHAWQGUC(host)
        testHAWQconfig(host)
        testDiskCapacity(host)
        testHDFSConfig(host)
        testIOSchedulers(host)
        testSysctl(host)
        testLimitsConf(host)
        testLinuxMounts(host)
        testNtp(host)


def testGenericSolarisHost(host):
    testSolarisEtcSystem(host)
    testSolarisEtcProject(host)
    testSolarisEtcUserAttr(host)

def testUnameConsistency():
    logger.info("test uname consistency")
    firstUname = None
    firstHost = None
    for _, host in gpcheck_info.hosts.items():
        uname = host.data.uname.output
        if firstUname:
            if firstUname != uname:
                checkFailed(h, "uname -r output different among hosts: %s : %s != %s : %s" % (firstHost, firstUname, host.hostname, uname))
        else:
            firstUname = uname
            firstHost = host.hostname


def testGenericLinuxCluster():
    for _, host in gpcheck_info.hosts.items():
        testGenericLinuxHost(host)
    testUnameConsistency()

def testGenericLinuxClusterBlockDev():
    for _, host in gpcheck_info.hosts.items():
        if not host.is_namenode:
            testBlockdev(host)

def testGenericSolarisCluster():
    for _, host in gpcheck_info.hosts.items():
        testGenericSolarisHost(host)
    testUnameConsistency()


def runTests():
    if gpcheck_info.host_type == HostType.GPCHECK_HOSTTYPE_GENERIC_LINUX:
        testGenericLinuxCluster()
        if gpcheck_info.is_root:
            testGenericLinuxClusterBlockDev()

    elif gpcheck_info.host_type == HostType.GPCHECK_HOSTTYPE_GENERIC_SOLARIS:
        testGenericSolarisCluster()

    else:
        raise GpCheckError("No tests exist for this platform in gpcheck")

    # report checks result
    logger.info("GPCHECK Result:")
    logger.info("---------------------------------------")
    if found_errors:
        logger.info("check failed!\tfound %s error(s)" % found_errors)
    else:
        logger.info("all check succeed!")
    logger.info("---------------------------------------")


def readZip():
    logger.info("trying to read zip file '%s'..." % options.zipin)

    words = options.zipin.split(".tar.gz")
    if len(words) != 2:
        raise GpCheckError("--zipin file needs to be a .tar.gz file")
    fname = words[0]

    # untar
    cmdStr = "tar xfz %s" % (options.zipin)
    if options.verbose:
        logger.info(cmdStr)
    try:
        cmd = Command("tarcmd", cmdStr)
        pool.addCommand(cmd)
        pool.join()
        items = pool.getCompletedItems()
        for i in items:
            if i.results.rc or i.results.halt or not i.results.completed:
                raise Exception("error running command '%s'" % cmdStr)

    except Exception, e:
        raise GpCheckError("Failed to extract tar file '%s': %s" % (options.zipin, e))

    # move extracted file to temp directory
    newfname = "%s/%s" % (tmpdir, fname)
    cmdStr = "mv %s %s" % (fname, newfname)
    if options.verbose:
        logger.info(cmdStr)
    try:
        cmd = Command("mvcmd", cmdStr)
        pool.addCommand(cmd)
        pool.join()
        items = pool.getCompletedItems()
        for i in items:
            if i.results.rc or i.results.halt or not i.results.completed:
                raise Exception("error running command '%s'" % cmdStr)

    except Exception, e:
        raise GpCheckError("Failed to move file '%s' to temp directory: %s" % (fname, e))

    # load pickle file
    global gpcheck_info
    try:
        with open(newfname, "rb") as f:
            gpcheck_info = pickle.load(f)

    except Exception, e:
        raise GpCheckError("Failed to load pickle file '%s': %s" % (newfname, e))

    logger.info("trying to read zip file '%s' [success]" % options.zipin)

def doZip(fname):
    logger.info("dump gpcheck data into a zip file '%s.tar.gz'..." % fname)

    # dump to pickle file
    try:
        with open(fname, "wb") as f:
            pickle.dump(gpcheck_info, f)

    except Exception, e:
        raise GpCheckError("Failed to dump pickle file '%s':\n%s" % (fname, e))

    # make a tar ball
    cmdStr = "tar cfz %s.tar.gz %s" % (fname, fname)
    if options.verbose:
        logger.info(cmdStr)
    try:
        cmd = Command("tarcmd", cmdStr)
        pool.addCommand(cmd)
        pool.join()
        items = pool.getCompletedItems()
        for i in items:
            if i.results.rc or i.results.halt or not i.results.completed:
                raise Exception("error running command '%s': %s" % (cmdStr, i.results.stderr.strip()))

    except Exception, e:
        raise GpCheckError("Failed to dump gpcheck data into a zip file:\n%s" % e)

    # delete pickle file
    cmdStr = "rm -rf %s" % fname
    if options.verbose:
        logger.info(cmdStr)
    try:
        cmd = Command("rmcmd", cmdStr)
        pool.addCommand(cmd)
        pool.join()
        items = pool.getCompletedItems()
        for i in items:
            if i.results.rc or i.results.halt or not i.results.completed:
                raise Exception("error running command '%s': %s" % (cmdStr, i.results.stderr.strip()))

    except Exception, e:
        raise GpCheckError("Failed to delete pickle file '%s':\n%s" % (fname, e))

    logger.info("dump gpcheck data into a zip file '%s.tar.gz' [success]" % fname)


def doPrint():
    for h in sorted(gpcheck_info.hosts):
        print "HOST: %s" % h
        print gpcheck_info.hosts[h].data
        print "----------------------------------------------------------------------\n"

    if gpcheck_info.hawq_collected_ok:
        print "HAWQ guc settings:"
        for guc_name, guc_val in gpcheck_info.hawq_gucs.items():
            print "GUC          : %s\nMaster  value: %s\nSegment value: %s\n" % (guc_name, guc_val[0], guc_val[1])

def checkHostsFile():
    '''
    This function will check /etc/hosts file, and if it contains IPv6, it will be commented out
    '''
    rtn_output = subprocess.check_output("cat /etc/hosts", shell=True)
    hostsContent = rtn_output.decode("utf8", "ignore")

    newHostsContent = ""
    # Trying to find if /etc/hosts contains ::1
    for hostName in hostsContent.splitlines():
        hostName = hostName.replace('\n', '')
        if hostName.find("::") > -1 and hostName.find("#") < 0:
            newHostsContent += "# " + hostName + "\n"
        else:
            newHostsContent += hostName + "\n"

    return newHostsContent

def local_run(cmd):
    '''Execute each shell command on local machine.'''
    for elem in cmd:
        result = subprocess.Popen(elem, shell=True).wait()
        check_return_code(result)

def checkHardwarePerf():
    """
    This function is used to check hardware performance of each node, which include disk IO, memory and network
    """
    global options
    hawq_home = os.getenv('GPHOME')
    if not hawq_home:
        print("HAWQ home directory is not defined, please check GPHOME settings.")
        sys.exit(1)
    source_hawq_env = ". %s/greenplum_path.sh" % hawq_home
    cmd = ""
    arguments = "-f %s -r dsN -D -S 8589934592 -d /tmp" % (options.file)
    cmd = "%s; gpcheckperf %s" % (source_hawq_env, arguments)
    rtn_output = subprocess.check_output(cmd, shell=True)
    rtnContent = rtn_output.decode("utf8", "ignore")

    # Display test result
    print(rtn_output)

    expected_disk_write_min_bandwidth = gpcheck_config.disk_write_min_bandwidth
    expected_disk_read_min_bandwidth = gpcheck_config.disk_read_min_bandwidth
    expected_stream_min_bandwidth = gpcheck_config.stream_min_bandwidth
    expected_network_min_bandwidth = gpcheck_config.network_min_bandwidth

    # Retrieve test result
    hostname_real_disk_write_min_bandwidth = ""
    real_disk_write_min_bandwidth = 0.0

    hostname_real_disk_read_min_bandwidth = ""
    real_disk_read_min_bandwidth = 0.0

    hostname_real_stream_min_bandwidth = ""
    real_stream_min_bandwidth = 0.0

    real_network_min_bandwidth = 0.0
    rtnContent = rtnContent.splitlines()
    for rowItem in rtnContent:
        if "disk write min bandwidth" in rowItem:
            value_hostnane = rowItem.split(": ")[1]
            hostname_real_disk_write_min_bandwidth = value_hostnane.split(" ")[1].strip()
            real_disk_write_min_bandwidth = (float)(value_hostnane.split(" ")[0].strip())
        elif "disk read min bandwidth" in rowItem:
            value_hostnane = rowItem.split(": ")[1]
            hostname_real_disk_read_min_bandwidth = value_hostnane.split(" ")[1].strip()
            real_disk_read_min_bandwidth = (float)(value_hostnane.split(" ")[0].strip())
        elif "stream min bandwidth" in rowItem:
            value_hostnane = rowItem.split(": ")[1]
            hostname_real_stream_min_bandwidth = value_hostnane.split(" ")[1].strip()
            real_stream_min_bandwidth = (float)(value_hostnane.split(" ")[0].strip())
        elif "min = " in rowItem:
            value_unit = rowItem.split("= ")[1]
            real_network_min_bandwidth = (float)(value_unit.split(" ")[0].strip())
        elif "[Warning] single host only - abandon netperf test" in rowItem:
            real_network_min_bandwidth = -1.0

    if real_disk_write_min_bandwidth >= expected_disk_write_min_bandwidth:
        logger.info("Pass: Disk write test")
    else:
        logger.error("Fail: Disk min write bandwidth %.2f (MB/s) is lower than expected value %.2f (MB/s). Node: %s." % (real_disk_write_min_bandwidth, expected_disk_write_min_bandwidth, hostname_real_disk_write_min_bandwidth)) 

    if real_disk_read_min_bandwidth >= expected_disk_read_min_bandwidth:
        logger.info("Pass: Disk read test")
    else:
        logger.error("Fail: Disk min read bandwidth %.2f (MB/s) is lower than expected value %.2f (MB/s). Node: %s." % (real_disk_read_min_bandwidth, expected_disk_read_min_bandwidth, hostname_real_disk_read_min_bandwidth))

    if real_stream_min_bandwidth >= expected_stream_min_bandwidth:
        logger.info("Pass: Stream test")
    else:
        logger.error("Fail: Stream min bandwidth %.2f (MB/s) is lower than expected value %.2f (MB/s). Node %s." % (real_stream_min_bandwidth, expected_stream_min_bandwidth, hostname_real_stream_min_bandwidth))

    if real_network_min_bandwidth < 0:
        logger.info("Warning: Single host only - abandon netperf test")
    elif real_network_min_bandwidth >= expected_network_min_bandwidth:
        logger.info("Pass: Network test")
    else:
        logger.error("Fail: Min network bandwidth %.2f (MB/s) is lower than expected value %.2f (MB/s)." % (real_network_min_bandwidth, expected_network_min_bandwidth))

def retrieve_DB_Version():
    '''
    This function is used to retrieve database version and RPM version.
    '''
    _sqlUtil = GpsqlUtil()
    db_version = _sqlUtil.performQuery("select version();")

    rtn_output = subprocess.check_output("sudo rpm -qa | grep hawq", shell=True)
    rpm_version = rtn_output.decode("utf8", "ignore")
    rpm_version = rpm_version.splitlines()[0].replace("\n", "")

    logger.info("Database Version is %s" % db_version[0]["version"])
    logger.info("RPM Version is %s" % rpm_version)

def retrieve_DB_process():
    '''
    This function is used to retrieve database process info.
    '''
    _sqlUtil = GpsqlUtil()
    db_process = _sqlUtil.performQuery("SELECT * FROM gp_segment_configuration;")

    if db_process is not None:
        logger.info("Database Process Information:")
        for listElem in db_process:
            if listElem["status"].lower() == "u":
                logger.info("Database node: %s is up." % listElem["hostname"])
            else:
                logger.error("Database node: %s is NOT up." % listElem["hostname"])
    else:
        logger.error("Fail to retrieve database process information.")

def check_DB_basic_function():
    '''
    This function will perform create table, insert data into table, select data from table 

    and drop table operations to test if database is functioning correctly.
    '''
    _sqlUtil = GpsqlUtil()

    table_name = "hawqTest_" + str(calendar.timegm(time.gmtime()))
    rowCount = 100000
    try:
        # Create table
        createTable = _sqlUtil.performQuery("CREATE TABLE %s (id INT);" % table_name, 0)
        if createTable != "FAIL":
            logger.info("Pass: Create table is successful.")
    except Exception as e:
        logger.error("Fail to create table: %s" % (str(e)))
        return None

    try:
        # Insert data into database
        insertData = _sqlUtil.performQuery("INSERT INTO %s SELECT generate_series(1, %s);" % (table_name, str(rowCount)), 0)
        rtnData = _sqlUtil.performQuery("SELECT count(*) FROM %s;" % (table_name))
        if rtnData[0]["count"] == rowCount:
            logger.info("Pass: Insert data into table is successful.")
            logger.info("Pass: Select data from table is successful.")
        else:
            logger.error("Fail to insert data into table.")
    except Exception as e:
        logger.error("Fail to insert data into table.")
        dropTable = _sqlUtil.performQuery("DROP TABLE %s;" % table_name)
        return None

    try:
        # Drop table
        dropTable = _sqlUtil.performQuery("DROP TABLE %s;" % table_name, 0)
        if dropTable != "FAIL":
            logger.info("Pass: Drop table is successful.")
    except Exception as e:
        logger.error("Fail to drop table %s. Please perform drop table operation manually." % table_name)
        return None

def check_GUC_configuration():
    '''
    This function will perform GUC configuration check:

    default_hash_table_bucket_number; hawq_rm_nvseg_perquery_limit; 

    hawq_rm_nvseg_perquery_perseg_limit;

    magma_shm_limit_per_block
    '''
    expected_default_hash_table_bucket_number = gpcheck_config.default_hash_table_bucket_number
    expected_hawq_rm_nvseg_perquery_limit = gpcheck_config.hawq_rm_nvseg_perquery_limit
    expected_hawq_rm_nvseg_perquery_perseg_limit = gpcheck_config.hawq_rm_nvseg_perquery_perseg_limit
    expected_magma_shm_limit_per_block = gpcheck_config.magma_shm_limit_per_block

    # Connect to DB to retrieve GUC config values
    _sqlUtil = GpsqlUtil()

    actual_default_hash_table_bucket_number = None
    actual_hawq_rm_nvseg_perquery_limit = None
    actual_hawq_rm_nvseg_perquery_perseg_limit =None
    actual_magma_shm_limit_per_block = None

    try:
        querySQL = "show default_hash_table_bucket_number;"
        actual_default_hash_table_bucket_number = _sqlUtil.performQuery(querySQL)
        actual_default_hash_table_bucket_number = actual_default_hash_table_bucket_number[0]["default_hash_table_bucket_number"]
        if expected_default_hash_table_bucket_number > int(actual_default_hash_table_bucket_number):
            logger.error("Fail: Default hash table bucket number %s is less than expected value %s." % (actual_default_hash_table_bucket_number, expected_default_hash_table_bucket_number))
        else:
            logger.info("Pass: Default hash table bucket number %s is NOT less than expected value %s." % (actual_default_hash_table_bucket_number, expected_default_hash_table_bucket_number))

        querySQL = "show hawq_rm_nvseg_perquery_limit;"
        actual_hawq_rm_nvseg_perquery_limit = _sqlUtil.performQuery(querySQL)
        actual_hawq_rm_nvseg_perquery_limit = actual_hawq_rm_nvseg_perquery_limit[0]["hawq_rm_nvseg_perquery_limit"]
        if expected_hawq_rm_nvseg_perquery_limit > int(actual_hawq_rm_nvseg_perquery_limit):
            logger.error("Fail: Hawq rm nvseg perquery limit %s is less than expected value %s." % (actual_hawq_rm_nvseg_perquery_limit, expected_hawq_rm_nvseg_perquery_limit))
        else:
            logger.info("Pass: Hawq rm nvseg perquery limit %s is NOT less than expected value %s." % (actual_hawq_rm_nvseg_perquery_limit, expected_hawq_rm_nvseg_perquery_limit))

        querySQL = "show hawq_rm_nvseg_perquery_perseg_limit;"
        actual_hawq_rm_nvseg_perquery_perseg_limit = _sqlUtil.performQuery(querySQL)
        actual_hawq_rm_nvseg_perquery_perseg_limit = actual_hawq_rm_nvseg_perquery_perseg_limit[0]["hawq_rm_nvseg_perquery_perseg_limit"]
        if expected_hawq_rm_nvseg_perquery_perseg_limit > int(actual_hawq_rm_nvseg_perquery_perseg_limit):
            logger.error("Fail: Hawq rm nvseg perquery perseg limit %s is less than expected value %s." % (actual_hawq_rm_nvseg_perquery_perseg_limit, expected_hawq_rm_nvseg_perquery_perseg_limit))
        else:
            logger.info("Pass: Hawq rm nvseg perquery perseg limit %s is NOT less than expected value %s." % (actual_hawq_rm_nvseg_perquery_perseg_limit, expected_hawq_rm_nvseg_perquery_perseg_limit))

        querySQL = "show magma_shm_limit_per_block;"
        actual_magma_shm_limit_per_block = _sqlUtil.performQuery(querySQL)
        actual_magma_shm_limit_per_block = actual_magma_shm_limit_per_block[0]["magma_shm_limit_per_block"].replace("MB", "")
        if expected_magma_shm_limit_per_block > int(actual_magma_shm_limit_per_block):
            logger.error("Fail: Magma shm limit per block %sMB is less than expected value %sMB." % (actual_magma_shm_limit_per_block, expected_magma_shm_limit_per_block))
        else:
            logger.info("Pass: Magma shm limit per block %sMB is NOT less than expected value %sMB." % (actual_magma_shm_limit_per_block, expected_magma_shm_limit_per_block))
    except Exception as e:
        logger.error("Fail to execute %s." % (querySQL))
        logger.error("Error Message: %s" % (str(e)))

def checkDiskUsage():
    '''
    This function is used to check disk total usage data of each node,

    and compare total disk space and available ratio with expected value in gpconfig.yml file
    '''
    global options

    rtn_output = subprocess.check_output("cat %s" % (options.file), shell=True)
    hostsContent = rtn_output.decode("utf8", "ignore")

    expected_disk_space_total = (float)(gpcheck_config.disk_space_total)
    expected_disk_space_available = (float)(gpcheck_config.diskusage_usagemax)

    # Check disk status of each host
    for elem in hostsContent.splitlines():
        hostName = elem.replace("\n", "")
        hostName = hostName.replace(" ", "")
        if hostName != "":
            cmd = []
            print("---------- Node Name: %s" % (hostName))
            cmd.append("ssh %s df -h --total | grep Filesystem; df -h --total | grep total" % (hostName))
            local_run(cmd)
            rtn_output = subprocess.check_output("ssh %s df -h -P --total -BG  | grep total" % (hostName), shell=True)
            rtnContent = rtn_output.decode("utf8", "ignore")
            rtnVal = ' '.join(rtnContent.splitlines()[0].split())
            rtnVal = rtnVal.split(" ")
            totalDiskSpace = (float)(rtnVal[1].replace("G", ""))
            diskAvailableSpace = (float)(rtnVal[3].replace("G", ""))
            diskAvaiableRatio = (diskAvailableSpace / totalDiskSpace) * 100.0

            # Compare real value with expected value
            if expected_disk_space_total > totalDiskSpace:
                logger.error("Fail: Disk total space %.1fGB is less than expected value %.1fGB." % (totalDiskSpace, expected_disk_space_total))
            else:
                logger.info("Pass: Disk total space %.1fGB is NOT less than expected value %.1fGB." % (totalDiskSpace, expected_disk_space_total))

            if expected_disk_space_available > diskAvaiableRatio:
                logger.error("Fail: Disk available space ratio %.1f%% is less than expected value %.1f%%." % (diskAvaiableRatio, expected_disk_space_available))
            else:
                logger.info("Pass: Disk available space ratio %.1f%% is NOT less than expected value %.1f%%." % (diskAvaiableRatio, expected_disk_space_available))

def checkMasterStandbySync():
    '''
    This function is used to check if master and standby are synchronized
    '''
    _sqlUtil = GpsqlUtil()
    gp_master_mirroring = _sqlUtil.performQuery("select * from gp_master_mirroring;")
    gp_master_mirroring_rtn = gp_master_mirroring[0]["summary_state"].lower()

    # Check if there is standby node
    gp_segment_configuration = _sqlUtil.performQuery("select * from gp_segment_configuration where role = 's';")
    standby_flag = False
    if len(gp_segment_configuration) > 0:
        standby_flag = True

    if standby_flag:
        if gp_master_mirroring_rtn == "synchronized":
            logger.info("Pass: Master and Standby are synchronized.")
        elif gp_master_mirroring_rtn == "not configured":
            logger.info("Fail: Standby is NOT found.")
        else:
            logger.error("Fail: Master and Standby are NOT synchronized.")
    else:
        if gp_master_mirroring_rtn == "synchronized":
            logger.info("Fail: Unexpected accident is occurring. Please check gp_segment_configuration and gp_master_mirroring tables.")
        elif gp_master_mirroring_rtn == "not configured":
            logger.info("Pass: Standby are NOT configured.")
        else:
            logger.error("Fail: Unexpected accident is occurring. Please check gp_segment_configuration and gp_master_mirroring tables.")

def checkTableAnalyzeStatus():
    '''
    This function is used to check if all tables are analyzed within three days
    '''
    _sqlUtil = GpsqlUtil()
    analyzeInfo = _sqlUtil.performQuery("SELECT classid, objid, staactionname, stasysid, stausename, stasubtype, statime FROM pg_stat_last_operation pslo, pg_class pc WHERE pc.oid = pslo.objid and (now()-statime)>3;")
    if len(analyzeInfo) > 0:
        logger.error("Fail: There are %d tables which are NOT analyzed within three days. Please check pg_stat_last_operation and pg_class tables to get details." % (len(analyzeInfo)))
    else:
        logger.info("Pass: All tables are analyzed within three days.")

def check_kerberos_enablement():
    '''
    Read /usr/local/hawq/etc/hdfs-client.xml and /usr/local/hawq/etc/yarn-client.xml to check if kerberos is enabled.
    '''
    # Open hdfs-client.xml and get data
    hdfs_client_Msg = subprocess.check_output("cat /usr/local/hawq/etc/hdfs-client.xml", shell=True)
    root = xml.etree.ElementTree.fromstring(hdfs_client_Msg)

    # Check if kerberos in hdfs-client.xml is enabled
    flag = True
    for elem in root.findall("property"):
        elemValue = elem.find('name').text
        if elemValue.lower() == "kerberos":
            flag = False
            break
    if flag:
        logger.error("Kerberos in hdfs-client.xml is NOT enabled.")
    else:
        logger.info("Kerberos in hdfs-client.xml is enabled.")

    # Open yarn-client.xml and get data
    yarn_client_Msg = subprocess.check_output("cat /usr/local/hawq/etc/yarn-client.xml", shell=True)
    root = xml.etree.ElementTree.fromstring(yarn_client_Msg)

    # Check if kerberos in yarn-client.xml is enabled
    flag = True
    for elem in root.findall("property"):
        elemValue = elem.find('name').text
        if elemValue.lower() == "kerberos":
            flag = False
            break
    if flag:
        logger.error("Kerberos in yarn-client.xml is NOT enabled.")
    else:
        logger.info("Kerberos in yarn-client.xml is enabled.")

def check_active_namenode():
    '''
    This function will read /usr/local/hawq/etc/hdfs-client.xml to check if first namenode is active namenode
    '''
    hdfs_client_Msg = subprocess.check_output("cat /usr/local/hawq/etc/hdfs-client.xml", shell=True)
    root = xml.etree.ElementTree.fromstring(hdfs_client_Msg)

    # Check Active Namenode in hdfs-client.xml
    flag = True
    firstNamenode = ""
    nameservices = ""
    for elem in root.findall("property"):
        if elem.find("name").text == "dfs.nameservices":
            nameservices = elem.find("value").text
            break

    for elem in root.findall("property"):
        if elem.find("name").text == "dfs.ha.namenodes." + nameservices:
            firstNamenode = elem.find("value").text.split(",")[0]
            flag = False
            break
    if flag:
        logger.error("Warning: Fail to retrieve namenode in hdfs-client.xml. Please check your HFDS configure in hdfs-client.xml.")
        return None

    try:
        # Check if first namenode is active
        subprocess.Popen("su - hdfs", shell=True).wait()
        command = "hdfs haadmin -getServiceState " + firstNamenode
        check_active_namenode_Msg = subprocess.check_output(command, shell=True)

        if check_active_namenode_Msg == "active":
            logger.info("Pass: First namenode %s is active name node." % (firstNamenode))
        elif check_active_namenode_Msg == "standby":
            logger.error("Fail: First namenode %s is NOT active name node." % (firstNamenode))
        else:
            logger.error("Fail: Fail to detect active name node.")
    except Exception as e:
        logger.error("Fail: HDFS Error: %s" % (str(e)))
    finally:
        # if current user is not root, switch back to root user
        current_user = subprocess.check_output("whoami", shell=True)
        if current_user != "root":
            subprocess.Popen("exit", shell=True).wait()

def check_hdfs_local_read():
    '''
    Read /usr/local/hawq/etc/hdfs-client.xml and /etc/hadoop/conf/hdfs-site.xml files and find <name>dfs.domain.socket.path</name>.

    If their socket file path are same, and permission is 666, it means hawq user can write them.
    '''
    # Check <name>dfs.domain.socket.path</name> in hdfs-client.xml
    hdfs_client_Msg = subprocess.check_output("cat /usr/local/hawq/etc/hdfs-client.xml", shell=True)
    root = xml.etree.ElementTree.fromstring(hdfs_client_Msg)

    flag = True
    socket_hdfs_client = ""
    for elem in root.findall("property"):
        elemValue = elem.find('name').text
        if elemValue.lower() == "dfs.domain.socket.path":
            socket_hdfs_client = elem.find("value").text
            flag = False
            break
    if flag:
        logger.error("Fail: Can Not find dfs.domain.socket.path in hdfs-client.xml.")
        return None

    # Check <name>dfs.domain.socket.path</name> in hdfs-site.xml
    hdfs_client_Msg = subprocess.check_output("cat /etc/hadoop/conf/hdfs-site.xml", shell=True)
    root = xml.etree.ElementTree.fromstring(hdfs_client_Msg)

    flag = True
    socket_hdfs_site = ""
    for elem in root.findall("property"):
        elemValue = elem.find('name').text
        if elemValue.lower() == "dfs.domain.socket.path":
            socket_hdfs_site = elem.find("value").text
            flag = False
            break
    if flag:
        logger.error("Fail: Can Not find dfs.domain.socket.path in hdfs-site.xml.")
        return None

    # Check file path
    if socket_hdfs_client == socket_hdfs_site:
        try:
            file_permission_Msg = subprocess.check_output("ls -l %s" % (socket_hdfs_site), shell=True)
            file_permission1 = file_permission_Msg[1:4]
            file_permission2 = file_permission_Msg[4:7]
            file_permission3 = file_permission_Msg[7:10]
            if file_permission1.find("-") < 0 or file_permission1.find("-") > 1:
                if file_permission2.find("-") < 0 or file_permission2.find("-") > 1:
                    if file_permission3.find("-") < 0 or file_permission3.find("-") > 1:
                        logger.info("Pass: HDFS local read is configured correctly.")
                    else:
                        logger.error("Fail: HDFS local read is NOT configured correctly.")
                else:
                    logger.error("Fail: HDFS local read is NOT configured correctly.")
            else:
                logger.error("Fail: HDFS local read is NOT configured correctly.")
        except Exception as e:
            logger.error("Fail: HDFS local read is NOT configured correctly. \n Error Message is: %s" % (e))
    else:
        logger.error("Fail: socket file path in %s and %s are different." % (socket_hdfs_client, socket_hdfs_site))

def check_return_code(result, logger = None,  error_msg = None, info_msg = None, exit_true = False):
    '''Check shell command exit code.'''
    if result != 0:
        if error_msg and logger:
            logger.error(error_msg)
    else:
        if info_msg and logger:
            logger.info(info_msg)
        if exit_true:
            sys.exit(0)
    return result

def overwrite_etc_hosts(filecontent = "", hostfile = ""):
    '''
    This function is used to overwrite /etc/hosts file and scp to all nodes in cluster
    '''
    cmd =[]
    hawq_home = os.getenv('GPHOME')
    if not hawq_home:
        print("HAWQ home directory is not defined, please check GPHOME settings.")
        sys.exit(1)
    source_hawq_env = ". %s/greenplum_path.sh" % hawq_home
    cmd.append(source_hawq_env)

    argument = "echo \'" + filecontent + "\' > /etc/hosts"
    cmd.append(argument)
    argument = "gpscp -f %s /etc/hosts =:/etc/hosts" % (hostfile)
    cmd.append(argument)

    local_run(cmd)

def enable_coredump(hostfile = ""):
    '''
    This function is used to enable core dump on each node
    '''
    cmd =[]
    hawq_home = os.getenv('GPHOME')
    if not hawq_home:
        print("HAWQ home directory is not defined, please check GPHOME settings.")
        sys.exit(1)
    source_hawq_env = ". %s/greenplum_path.sh" % hawq_home
    cmd.append(source_hawq_env)

    argument = "gpssh -f %s ulimit -c unlimited" % (hostfile)
    cmd.append(argument)

    local_run(cmd)

def install_ntp(hostfile = ""):
    '''
    This function is used to install and enable ntp on each node
    '''
    cmd =[]
    hawq_home = os.getenv('GPHOME')
    if not hawq_home:
        print("HAWQ home directory is not defined, please check GPHOME settings.")
        sys.exit(1)
    source_hawq_env = ". %s/greenplum_path.sh" % hawq_home
    cmd.append(source_hawq_env)

    argument = "gpssh -f %s -e 'yum install -y ntp'" % (hostfile)
    cmd.append(argument)
    argument = "gpssh -f %s -e 'systemctl enable ntpd'" % (hostfile)
    cmd.append(argument)
    argument = "gpssh -f %s -e 'systemctl start ntpd'" % (hostfile)
    cmd.append(argument)

    local_run(cmd)

def disable_firewall(hostfile = ""):
    '''
    This function is used to disable firewall on each node
    '''
    cmd =[]
    hawq_home = os.getenv('GPHOME')
    if not hawq_home:
        print("HAWQ home directory is not defined, please check GPHOME settings.")
        sys.exit(1)
    source_hawq_env = ". %s/greenplum_path.sh" % hawq_home
    cmd.append(source_hawq_env)

    argument = "gpssh -f %s -e 'systemctl stop iptables'" % (hostfile)
    cmd.append(argument)
    argument = "gpssh -f %s -e 'systemctl disable iptables'" % (hostfile)
    cmd.append(argument)
    argument = "gpssh -f %s -e 'systemctl stop firewalld'" % (hostfile)
    cmd.append(argument)
    argument = "gpssh -f %s -e 'systemctl disable firewalld'" % (hostfile)
    cmd.append(argument)

    local_run(cmd)

def disable_selinux(hostfile = ""):
    '''
    This function is used to disable selinux on each node
    '''
    cmd =[]
    hawq_home = os.getenv('GPHOME')
    if not hawq_home:
        print("HAWQ home directory is not defined, please check GPHOME settings.")
        sys.exit(1)
    source_hawq_env = ". %s/greenplum_path.sh" % hawq_home
    cmd.append(source_hawq_env)

    argument = "gpssh -f %s -e 'sed -i \"s/^SELINUX\=enforcing/SELINUX\=disabled/g\" /etc/selinux/config'" % (hostfile)
    cmd.append(argument)
    argument = "gpssh -f %s -e 'setenforce 0'" % (hostfile)
    cmd.append(argument)

    local_run(cmd)

function_dispatcher = {} # Include all hawq check --postSetup --postUpgrade and --weeklyExamine functions
function_dispatcher.update({"overwrite_etc_hosts" : overwrite_etc_hosts})
function_dispatcher.update({"enable_coredump" : enable_coredump})
function_dispatcher.update({"install_ntp" : install_ntp})
function_dispatcher.update({"disable_firewall" : disable_firewall})
function_dispatcher.update({"disable_selinux" : disable_selinux})
function_dispatcher.update({"checkHardwarePerf" : checkHardwarePerf})
function_dispatcher.update({"Check config in /etc/sysctl.conf && /etc/security/limits.conf" : runTests})
function_dispatcher.update({"retrieve_DB_Version" : retrieve_DB_Version})
function_dispatcher.update({"retrieve_DB_process" : retrieve_DB_process})
function_dispatcher.update({"check_DB_basic_function" : check_DB_basic_function})
function_dispatcher.update({"check_GUC_configuration" : check_GUC_configuration})
function_dispatcher.update({"check_kerberos_enablement" : check_kerberos_enablement})
function_dispatcher.update({"check_active_namenode" : check_active_namenode})
function_dispatcher.update({"check_hdfs_local_read" : check_hdfs_local_read})
function_dispatcher.update({"checkDiskUsage" : checkDiskUsage})
function_dispatcher.update({"checkMasterStandbySync" : checkMasterStandbySync})
function_dispatcher.update({"checkTableAnalyzeStatus" : checkTableAnalyzeStatus})

if __name__ == '__main__':

    if gpcheck_info.is_root:
        logger.info("gpcheck will perform block device's readahead checks when run as root")
    try:
        try:
            checkPlatform()
            parseargs()
            readConfigFile()

        except GpCheckError, e:
            logger.error(str(e))
            sys.exit(1)
            if pool:
                pool.join()
                pool.haltWork()
                pool.joinWorkers()

        try:
            tmpdir = tempfile.mkdtemp(prefix='gpcheck')
        except Exception, e:
            logger.error("Error creating tmp dir on master: %s" % e)
            sys.exit(1)

        try:
            # Phase 1: collect input
            if options.zipin:
                readZip() # load information into gpcheck_info from zip
            else:
                # read host info into gpcheck_info.hosts from --file or --host
                createHostList()
                # collect each server's system environment configuration
                runCollections()
                # read collected data into gpcheck_info
                readDataFiles()
                # read HAWQ configuration
                readHAWQConfiguration()

            # Phase 2: generate output
            if options.postSetup:
                cmd = []
                counter = 1

                logger.info("-------------- Execute post-setup check --------------\n")

                # Step 1: Comment out ::1 localhost in /etc/hosts
                logger.info("-------------- Step %d: Comment out ::1 localhost in /etc/hosts" % counter)
                file_content = checkHostsFile()
                function_dispatcher["overwrite_etc_hosts"](file_content, options.file)
                counter += 1
                print("")

                # Step 2: Check hardware performance
                logger.info("-------------- Step %d: Check hardware performance (This step will take a little bit more time to perform)" % counter)
                function_dispatcher["checkHardwarePerf"]()
                counter += 1
                print("")

                # Step 3: Enable core dump
                logger.info("-------------- Step %d: Enable core dump" % counter)
                function_dispatcher["enable_coredump"](options.file)
                counter += 1
                print("")

                # Step 4: Install NTP
                logger.info("-------------- Step %d: Install NTP" % counter)
                function_dispatcher["install_ntp"](options.file)
                counter += 1
                print("")

                # Step 5: Disable Firewall
                logger.info("-------------- Step %d: Disable Firewall" % counter)
                function_dispatcher["disable_firewall"](options.file)
                counter += 1
                print("")

                # Step 6: Disable Selinux
                logger.info("-------------- Step %d: Disable Selinux" % counter)
                function_dispatcher["disable_selinux"](options.file)
                counter += 1
                print("")

                # Step 7: Check configuration in /etc/sysctl.conf and /etc/security/limits.conf
                logger.info("-------------- Step %d: Check configuration in /etc/sysctl.conf and /etc/security/limits.conf" % counter)
                function_dispatcher["Check config in /etc/sysctl.conf && /etc/security/limits.conf"]()
                counter += 1
                print("")

                # Step 8: Retrieve Database Version
                logger.info("-------------- Step %d: Retrieve Database Version" % counter)
                function_dispatcher["retrieve_DB_Version"]()
                counter += 1
                print("")

                # Step 9: Retrieve Database Process
                logger.info("-------------- Step %d: Retrieve Database Process" % counter)
                function_dispatcher["retrieve_DB_process"]()
                counter += 1
                print("")

                # Step 10: Check Database Basic Functions
                logger.info("-------------- Step %d: Check Database Basic Functions" % counter)
                function_dispatcher["check_DB_basic_function"]()
                counter += 1
                print("")

                # Step 11: Check GUC Configuration
                logger.info("-------------- Step %d: Check GUC Configuration" % counter)
                function_dispatcher["check_GUC_configuration"]()
                counter += 1
                print("")

                # Step 12: Check Kerberos Enablement
                logger.info("-------------- Step %d: Check Kerberos Enablement" % counter)
                function_dispatcher["check_kerberos_enablement"]()
                counter += 1
                print("")

                # Step 13: Check Active Namenode
                logger.info("-------------- Step %d: Check Active Namenode" % counter)
                function_dispatcher["check_active_namenode"]()
                counter += 1
                print("")

                # Step 14: Check HDFS Local Read
                logger.info("-------------- Step %d: Check HDFS Local Read" % counter)
                function_dispatcher["check_hdfs_local_read"]()
                counter += 1
                print("")
            elif options.postUpgrade:
                counter = 1

                logger.info("-------------- Execute post-upgrade check --------------\n")

                # Step 1: Retrieve Database Process
                logger.info("-------------- Step %d: Retrieve Database Process" % counter)
                function_dispatcher["retrieve_DB_process"]()
                counter += 1
                print("")

                # Step 2: Retrieve Database Version
                logger.info("-------------- Step %d: Retrieve Database Version" % counter)
                function_dispatcher["retrieve_DB_Version"]()
                counter += 1
                print("")

                # Step 10: Check Database Basic Functions
                logger.info("-------------- Step %d: Check Database Basic Functions" % counter)
                function_dispatcher["check_DB_basic_function"]()
                counter += 1
                print("")
            elif options.weeklyExamine:
                cmd = []
                counter = 1

                logger.info("-------------- Execute weekly-examine check --------------\n")

                # Step 1: Check Disk Usage
                logger.info("-------------- Step %d: Check Disk Usage" % counter)
                function_dispatcher["checkDiskUsage"]()
                counter += 1
                print("")

                # Step 2: Check if Master and Standby are synchronized
                logger.info("-------------- Step %d: Check if Master and Standby are synchronized" % counter)
                function_dispatcher["checkMasterStandbySync"]()
                counter += 1
                print("")

                # Step 3: Check if the specified table is analyzed within three days
                logger.info("-------------- Step %d: Check if the specified table is analyzed within three days" % counter)
                function_dispatcher["checkTableAnalyzeStatus"]()
                counter += 1
                print("")
            elif options.stdout:
                doPrint()
            elif options.zipout:
                doZip("./gpcheck_%s" % time.time())
            else:
                runTests()
                if found_errors:
                    sys.exit(1)

        except GpCheckError, e:
            logger.error(str(e))
            sys.exit(1)

        finally:
            logger.info("Clean up...")
            try:
                if tmpdir:
                    shutil.rmtree(tmpdir)
            except Exception, e:
                logger.error("error removing tempdir during job cleanup: %s" % e)
    finally:
        if pool:
            pool.join()
            pool.haltWork()
            pool.joinWorkers()