怎么做 HDFS 的原地平滑缩容？

你的名字 2022-11-22 00:12 166阅读 0赞

## 背景 ##

目前有三个DataNode节点,需要缩容这三个DataNode节点上的磁盘. 即修改`dfs.data.dir中指定的目录...`

## 三种缩容方案 ##

第一种缩容方法，如果 DataNode 的节点数目比较多，并且允许缩减存储空间的同时缩减 CPU 和内存资源，则可以缩掉若干个 DataNode 节点，直接使用 HDFS 提供的 decommission（附上 [社区版文档][Link 1]，[CDH 文档][CDH]，[HDP 文档][HDP]）。这是最常见的方法，缩减过程中涉及大量数据的跨节点迁移，会产生大量的内网流量可能影响线上负载，需要运维人员保持密切关注和手动调优，通常需要一两周的时间。如果集群只剩 3 个 DataNode 节点，或者上面的 CPU 或者 内存资源不能同步缩减时，就不能用这个方法了。

第二种缩容方法，即在保持 DataNode 节点数不变的情况下，缩减每个节点上的磁盘空间，可以修改 DataNode 上的 `dfs.data.dir`参数，删掉一个或者多个磁盘目录，然后等待 HDFS 自动补充副本。这个方法统一也会导致节点间的大量数据移动，会产生大量的内网流量可能影响线上负载，需要运维人员保持密切关注和手动调优，可能也需要一两周时间。此外，如果数据只有 2 副本，相对会比较危险，一旦删除一个磁盘目录时正好有节点出问题或者某块磁盘坏掉，极有可能造成数据缺失。

以上两种方法都会产生大量网络流量，可能影响线上服务，并且会增加丢失数据的风险。本文提供第三种方法，怎么在缩容的同时，尽量避免产生的内网流量影响线上工作负载，同时尽量减少缩容过程中数据丢失的风险。

## 方案分析 ##

首先我们看一下 DataNode 在磁盘上的目录结构：

└── dn
        ├── current
        │   ├── BP-847673977-192.168.0.120-1559552771699
        │   │   ├── current
        │   │   │   ├── dfsUsed
        │   │   │   ├── finalized
        │   │   │   │   ├── subdir0
        │   │   │   │   │   ├── subdir1
        │   │   │   │   │   │   ├── blk_1073742303
        │   │   │   │   │   │   ├── blk_1073742303_1479.meta
        │   │   │   ├── rbw
        │   │   │   └── VERSION
        │   │   ├── scanner.cursor
        │   │   └── tmp
        │   └── VERSION
        └── in_use.lock

*  `BP-847673977-192.168.0.120-1559552771699`：这是块池目录，如果以 Federation 方式部署的时候，会有多个块池目录。
 *  `dfsUsed`：保存的是磁盘的使用统计数据，每 10 分钟刷新一次。
 *  `finalized 和 rbw 目录：这两个`都是用于存储数据块的，`finalized` 放的是已经完成写入的数据块，`rbw` 是正在写入的数据块。每个数据块对应 2 个文件，`blk` 文件存放数据，另外一个以 `meta` 结尾的存放校验和等元数据。
 *  `VERSION` 文件：主要包含布局版本、集群 ID、DataNode ID、块池 ID 等信息。
 *  `scanner.cursor 文件`：DataNode 会定期的对每个 `blk` 文件做校验，这个文件是用来记录校验到哪个位置的。

不难看出所有的数据文件都存在 `finalized` 和 `rbw` 里面，并且同一个 DataNode 上面不会存在相同 Block ID 的数据文件。因此完全可以通过迁移 `blk` 文件的方式来将一块磁盘上面的数据移动到另外一块磁盘上，然后在卸载此磁盘来达到缩容的目的。

## 缩容步骤 ##

本文示例的 HDFS 是 CDH 5.16 版本，使用 ClouderaManager 管理集群。集群只有 3 个节点，每个节点有多块 SSD 盘，数据两副本，存储利用率很低，每个节点都可以卸载掉一块磁盘，但是无法使用前面两种常见的缩容方法，同时缩容过程要尽可能可能减小对线上服务的影响。

**以下操作均是针对单一 DataNode 的操作，其他 DataNode 也需要按照以下步骤执行（可以适当并行）**：

1.  选择磁盘。选择需要被卸载的数据盘和接收数据的数据盘，注意要确保接收数据的磁盘剩余空间够大于被卸载磁盘上的数据。这里假设：
    
    被卸载磁盘：`/dfs1`，此磁盘上的 DataNode 数据目录：`/dfs1/dfs/dn`
    
    数据接收盘：`/dfs`，此磁盘上的 DataNode 数据目录：`/dfs/dfs/dn`
2.  第一次复制数据。从 `dfs.data.dir` 里面挑选出在被卸载磁盘上面的目录，然后将此目录全量拷贝到接收数据盘上，为了尽量减少对 IO 的占用，用 `ionice` 加 `rsync` 的方式拷贝数据，确保不阻塞高优先级的任务。
    
        ionice -c 2 -n 7 rsync -au /dfs1/dfs/dn/ /dfs/shrink_temp/dn
3.  需要保证数据都被拷贝过去了，所以需要将 DataNode 停掉。可以通过 ClouderaManager 界面关闭 DataNode。
4.  第二次增量复制数据。重复步骤 2，将在步骤 2 和步骤 3 之间的新增数据增量更新到接收盘上。增量数据会比较少，估计能很快完成。
    
        ionice -c 2 -n 7 rsync -au /dfs1/dfs/dn/ /dfs/shrink_temp/dn
5.  合并目录。此时被卸载盘上的数据都已经复制到接收盘上，但是数据还在原来的文件夹里面。**如果同一块磁盘上面有两个 DataNode 数据目录的话，会导致 HDFS 容量计算重复，因此需要合并。**可以通过 `rsync` 的硬链的方式将数据拷贝过去，这样不涉及真正的数据拷贝，执行非常快，同时将拷贝过去的源数据删除。检查剩余数据是否有 `blk` 文件，没有就合并完成。
    
        ionice -c 2 -n 7 rsync -au --link-dest=/dfs/shrink_temp/dn --ignore-existing --remove-source-files /dfs/shrink_temp/dn/ /dfs/dfs/dn
6.  通过 ClouderaManager 修改 `dfs.data.dir` 配置项，删除卸载磁盘上的数据目录。
7.  通过 ClouderaManager 启动 DataNode，并检查 HDFS 的状态。
    
        sudo -u hdfs hdfs fsck /

**为什么不直接将被卸载盘的数据复制合并到接收盘的 DataNode 数据目录里面呢？** 这是因为，在第一次拷贝的时候，DataNode 仍然在运行，因此 DataNode 会定期检查副本数量，此时拷贝过去的数据算是额外副本，有可能会被 DataNode 删除掉。

整个缩容过程中 DataNode 停止的时间只是步骤 4 和步骤 5 所需时间。步骤 4 是增量拷贝，速度很快，步骤 5 只是文件元数据操作，同样很快。

以上步骤看起来比较多，手动操作会容易出错，因此我们将以上缩容过程写了一个脚本（部分操作依赖 Hadoop 发行版的 API，目前支持 CDH5），请下载[setup-hadoop.py][]，运行命令，并按照提示输入进行缩容：

python setup-hadoop.py shrink_datanode

#!/usr/bin/env python
    # coding=utf-8
    import base64
    import getpass
    import glob
    import json
    import os
    import platform
    import re
    import shutil
    import socket
    import stat
    import sys
    import time
    import xml.etree.ElementTree as ET
    from email.utils import parsedate_tz, mktime_tz
    from functools import wraps
    from os.path import getsize, join
    
    try:
        raw_input = raw_input
    except NameError:
        raw_input = input
    
    try:
        from collections import OrderedDict as dict
    except ImportError:
        pass
    
    try:
        import commands
    except ImportError:
        import subprocess as commands
    
    try:
        from urllib.parse import urlparse, urlencode, urljoin
        from urllib.request import urlopen, Request
        from urllib.error import HTTPError
    except ImportError:
        from urlparse import urlparse, urljoin
        from urllib import urlencode
        from urllib2 import urlopen, Request, HTTPError
    
    try:
        import configparser
    except ImportError:
        import ConfigParser as configparser
    
    OS = platform.system()
    
    VERSION = os.environ.get("JFS_VERSION", "0.7-beta")
    JAR_FULL_NAME = "juicefs-hadoop-" + str(VERSION) + ".jar"
    JAR_NAME = "juicefs-hadoop.jar"
    
    
    def log(msg):
        sys.stderr.write(str(msg) + "\n")
    
    
    JAVA8_HOME_CANDIDATES = [
        '/usr/jdk64/jdk1.8',
        '/usr/java/jdk1.8',
        '/usr/lib/jvm/j2sdk1.8-oracle',
        '/usr/lib/jvm/java-8-oracle',
        '/usr/lib/jdk8-latest',
        '/usr/lib/jvm/java-1.8.0',
        '/usr/lib/jvm/java-1.8.0-oracle',
        '/home/sa_cluster/sa/java',
        '/usr/local/jdk1.8'
    ]
    
    OPENJAVA8_HOME_CANDIDATES = [
        '/usr/lib/jvm/java-1.8.0-openjdk-amd64',
        '/usr/lib/jvm/java-1.8.0-openjdk-ppc64el',
        '/usr/lib/jvm/java-1.8.0-openjdk',
        '/usr/lib64/jvm/java-1.8.0-openjdk-1.8.0',
    ]
    
    MISCJAVA_HOME_CANDIDATES = [
        '/Library/Java/Home',
        '/usr/java/default',
        '/usr/lib/jvm/java',
        '/usr/lib/jvm/default-java',
        '/usr/lib/jvm/java-openjdk',
        '/usr/lib/jvm/java-oracle',
    ]
    
    JAVA8_HOME_CANDIDATES.extend(OPENJAVA8_HOME_CANDIDATES)
    JAVA8_HOME_CANDIDATES.extend(MISCJAVA_HOME_CANDIDATES)
    
    TEMP_DIR = '/tmp/juicefs-tmp'
    
    HADOOP_CMD = os.getenv('HADOOP_COMMAND')
    if not HADOOP_CMD:
        sts, HADOOP_CMD = commands.getstatusoutput('which hadoop')
        if sts > 0:
            log("You are not running this script in hadoop env")
            HADOOP_CMD = ""
            # sys.exit(1)
    SSH = 'ssh -t -o StrictHostKeyChecking=no'
    
    
    def detect_java_home():
        if os.getenv('JAVA_HOME') is None:
            for candidate_regex in JAVA8_HOME_CANDIDATES:
                for candidate in glob.glob(candidate_regex + "*"):
                    if os.path.exists(os.path.join(candidate, "bin/java")):
                        return candidate
            log("JAVA_HOME auto detect failed! You should confirm it first!")
            sys.exit(1)
        else:
            return os.getenv('JAVA_HOME')
    
    
    JAVE_HOME = detect_java_home()
    
    
    def get_jar_url():
        base = "https://s.juicefs.com/static"
        jar_url = base + "/%s/" % platform.system()
        jar_url += JAR_FULL_NAME
        return jar_url
    
    
    def remove_jar(lib_dir):
        jars = glob.glob(os.path.join(lib_dir, "juicefs-hadoop*.jar"))
        for jar in jars:
            if os.path.exists(jar):
                log("removing %s" % jar)
            os.remove(jar)
    
    
    def retry(tries=6, delay=1, backoff=2):
        def deco_retry(f):
            @wraps(f)
            def f_retry(*args, **kwargs):
                mtries, mdelay = tries, delay
                while mtries > 0:
                    try:
                        return f(*args, **kwargs)
                    except Exception as e:
                        msg = 'func %s raises %s, retrying in %d seconds...' % (f.__name__, e, mdelay)
                        log(msg)
                        time.sleep(mdelay)
                        mtries -= 1
                        mdelay *= backoff
                log("func %s failed %d times, abort" % (f.__name__, tries))
                sys.exit(1)
    
            return f_retry
    
        return deco_retry
    
    
    @retry(tries=3)
    def download(path, url, mode=stat.S_IXUSR | stat.S_IRUSR | stat.S_IROTH | stat.S_IXOTH | stat.S_IRGRP | stat.S_IXGRP):
        request = Request(url)
        request.add_header('Accept-encoding', 'gzip')
        request.add_header('User-Agent', 'JuiceFS')
    
        response = urlopen(request)
        assert response.getcode() == 200
        data = response.read()
        last_mod = parse_last_modified(response.info().get('last-modified'))
    
        tmp = path + ".tmp"
        d = os.path.dirname(tmp)
        if not os.path.exists(d):
            os.makedirs(d, mode | stat.S_IWUSR)
        with open(tmp, 'wb') as f:
            f.write(data)
        os.chmod(tmp, mode)
    
        if last_mod:
            atime = os.path.getatime(tmp)
            os.utime(tmp, (atime, last_mod))
    
        os.rename(tmp, path)
        mtime = os.path.getmtime(path)
        return not last_mod or last_mod <= mtime
    
    
    def parse_last_modified(header):
        mod = parsedate_tz(header)
        if mod:
            return mktime_tz(mod)
    
    
    def in_localhost(host):
        ip_lists = socket.gethostbyname_ex(socket.gethostname())
        if host == ip_lists[0] or host in ip_lists[1] or host in ip_lists[2]:
            return True
        return False
    
    
    JUICEFS_CONF = dict([
        ('fs.jfs.impl', 'com.juicefs.JuiceFileSystem'),
        ('fs.AbstractFileSystem.jfs.impl', 'com.juicefs.JuiceFS'),
        ('juicefs.cache-dir', '/var/jfsCache'),
        ('juicefs.cache-size', '0'),
        ('juicefs.cache-group', 'yarn'),
        ('juicefs.discover-nodes-url', 'yarn'),
        ('juicefs.access-log', '/tmp/juicefs.access.log'),
        ('juicefs.conf-dir', '/etc/juicefs')
    ])
    
    
    def is_jfs_conf(key):
        if key.startswith('juicefs'):
            return True
        if key in ['fs.jfs.impl', 'fs.AbstractFileSystem.jfs.impl']:
            return True
    
    
    def get_conf_path():
        root_conf_path = '/root/.juicefs'
        default_conf_path = os.path.join(JUICEFS_CONF['juicefs.conf-dir'], '.juicefs')
        if not os.path.exists(root_conf_path) and not os.path.exists(default_conf_path):
            log('you should run \"sudo python setup-hadoop.py install\" first')
            sys.exit(1)
        return root_conf_path if os.path.exists(root_conf_path) else default_conf_path
    
    
    JFS_NAME = None
    
    
    def get_jfs_name():
        jfs_name = os.getenv('JUICEFS_NAME')
        if jfs_name:
            return jfs_name
        global JFS_NAME
        if JFS_NAME:
            return JFS_NAME
        conf_path = get_conf_path()
        conf_file_pattern = os.path.join(conf_path, '*.conf')
        files = glob.glob(conf_file_pattern)
        if len(files) < 1:
            log("No config file was found, please mount JuiceFS first!")
            sys.exit(1)
        names = [os.path.basename(x)[:-5] for x in files]
        if len(files) > 1:
            name = raw_input("please input your JuiceFS Name(%s):" % ", ".join(names)) or names[0]
        else:
            name = names[0]
        JFS_NAME = name
        return name
    
    
    def has_jfs_conf(key, dic):
        jfs_name = get_jfs_name()
        items = key.split('.')
        k = key.replace('juicefs.%s.' % jfs_name, 'juicefs.') \
            if jfs_name == items[1] else key.replace('juicefs.', 'juicefs.%s.' % jfs_name)
        return (key in dic.keys()) or (k in dic.keys())
    
    
    def check_input(username, password, url):
        request = Request(url, data=None)
        base64string = base64.b64encode('%s:%s' % (username, password))
        request.add_header("Authorization", "Basic %s" % base64string)
        try:
            urlopen(request)
        except:
            log("\nError: url or username or password is wrong, please check\n\tcurrent url is %s" % url)
            sys.exit(1)
    
    
    def do_request(url, username, password, request_type, data='', header=None):
        if header is None:
            header = {}
        request = Request(url)
        for key, value in header.items():
            request.add_header(key, value)
        base64string = base64.b64encode('%s:%s' % (username, password))
        request.add_header("Authorization", "Basic %s" % base64string)
        request.add_data(data)
        request.get_method = lambda: request_type
        return urlopen(request)
    
    
    def list_all_xml_conf(content):
        """
        :param content:
        :return: OrderedDict()
        """
        properties = ET.fromstring(content)
        result = dict()
        for i in list(properties):
            if i.find("name") is not None:
                result[i.find("name").text.strip()] = i.find("value").text if i.find("value").text is not None else ''
        return result
    
    
    def read_xml_conf(content, key):
        """
        :param content:
        :return: value
        """
        conf = list_all_xml_conf(content)
        return conf.get(key, '')
    
    
    def get_mount_point(pathname):
        """Get the mount point of the filesystem containing pathname"""
        pathname = os.path.normcase(os.path.realpath(pathname))
        parent_device = path_device = os.stat(pathname).st_dev
        while parent_device == path_device:
            mount_point = pathname
            pathname = os.path.dirname(pathname)
            if pathname == mount_point:
                break
            parent_device = os.stat(pathname).st_dev
        return mount_point
    
    
    def get_mounted_device(pathname):
        """Get the device mounted at pathname"""
        # uses "/proc/mounts"
        mount_point = get_mount_point(pathname)  # might be unnecessary here
        try:
            with open("/proc/mounts", "r") as ifp:
                for line in ifp:
                    fields = line.rstrip('\n').split()
                    # note that line above assumes that
                    # no mount points contain whitespace
                    if fields[1] == mount_point:
                        return fields[0]
        except EnvironmentError:
            pass
        return None  # explicit
    
    
    def get_fs_freespace(pathname):
        """Get the free space of the filesystem containing pathname"""
        stat = os.statvfs(pathname)
        return stat.f_bfree * stat.f_bsize
    
    
    def get_dir_size(dir):
        size = 0
        for root, dirs, files in os.walk(dir):
            size += sum([getsize(join(root, name)) for name in files])
        return size
    
    
    def exec_cmd(cmd, err_msg=None):
        code, text = commands.getstatusoutput(cmd)
        if code > 0:
            if err_msg is not None:
                log(err_msg)
            log("exec command: %s due to: %s" % (cmd, str(text)))
        return code >> 8
    
    
    def rsync_data(src, dst, hard_link=False, ignore_existing=False, remove_source_files=False, files_from=None):
        if not os.path.exists(src):
            return
        start = time.time()
        if os.path.isdir(src):
            src = os.path.abspath(src) + "/"
        if not os.path.exists(dst):
            os.makedirs(dst)
        rsync = "rsync -au"
        if files_from:
            src = "/"
            rsync += " --recursive --files-from=%s " % files_from
        if hard_link:
            rsync += " --link-dest=%s" % os.path.abspath(src)
        if ignore_existing:
            rsync += " --ignore-existing"
        if remove_source_files:
            rsync += " --remove-source-files"
        cmd = 'ionice -c 2 -n 7 %s %s %s' % (rsync, src, dst)
        log("rsyncing data from %s to %s, command: %s" % (src, dst, cmd))
        code = exec_cmd(cmd)
        if code > 0 and code != 24:
            log('shrink datanode on %s failed' % socket.gethostname())
            sys.exit(1)
        end = time.time()
        log('rsync success, cost time %ds' % (end - start))
    
    
    def extract_number(text, prefix):
        pattern = re.compile(r'(?<=%s)\d+\.?\d*' % prefix)
        return pattern.findall(text)[0]
    
    
    class DataNode:
        def __init__(self, host, group, data_dir):
            self.host = host
            self.group = group
            self.data_dir = data_dir
    
        def __str__(self):
            return "host: %s, group: %s, data_dir: %s" % (self.host, self.group, self.data_dir)
    
    
    class HadoopDist(object):
        def __init__(self):
            self.download_dir = "/usr/local/lib"
            self.link_dirs = []
            self.need_to_add_conf = dict()
            self.current_conf = dict()
            self.core_site_path = "/etc/hadoop/conf/core-site.xml"
            self.hdfs_site_path = "/etc/hadoop/conf/hdfs-site.xml"
            self.mapred_site_path = "/etc/hadoop/conf/mapred-site.xml"
            self.hdfs_superuser = 'hdfs'
    
        def parse_conf(self, config_file):
            log("loading configurations from file: %s" % config_file)
            config_dic = json.load(open(config_file))
            result = JUICEFS_CONF.copy()
    
            result["juicefs.token"] = config_dic['token']
            if 'accesskey' in config_dic.keys():
                accesskey = config_dic['accesskey']
                if accesskey:
                    result['juicefs.accesskey'] = accesskey
            if 'secretkey' in config_dic.keys():
                secretkey = config_dic['secretkey']
                if secretkey:
                    result['juicefs.secretkey'] = secretkey
    
            return result
    
        def ln_jar(self, lib_dir, use_tools_jar=True):
            if lib_dir is not None and os.path.exists(lib_dir):
                log("installing juicefs-hadoop.jar to dir: %s" % lib_dir)
                juicefs_jar = os.path.join(lib_dir, JAR_NAME)
                if not os.path.exists(juicefs_jar):
                    os.symlink(self.jar_link_path, juicefs_jar)
                if use_tools_jar:
                    tools_jar = os.path.join(lib_dir, "tools.jar")
                    if os.path.islink(tools_jar) or os.path.exists(tools_jar):
                        os.remove(tools_jar)
                    os.symlink(os.path.join(JAVE_HOME, "lib/tools.jar"), tools_jar)
    
        @property
        def jar_link_path(self):
            return os.path.join(self.download_dir, JAR_NAME)
    
        @property
        def install_path(self):
            return os.path.join(self.download_dir, JAR_FULL_NAME)
    
        def download_jar(self):
            if os.path.exists(self.install_path):
                os.remove(self.install_path)
            if os.path.islink(self.jar_link_path):
                os.remove(self.jar_link_path)
            download_jar = os.environ.get("DOWNLOAD_JUICEFS", 'true')
            if download_jar.lower() == 'true':
                url = get_jar_url()
                log("downloading %s from %s to %s" % (JAR_FULL_NAME, url, self.download_dir))
                download(self.install_path, url)
            else:
                jar = os.path.join(TEMP_DIR, JAR_FULL_NAME)
                if os.path.exists(jar):
                    os.system("mv -i %s %s" % (jar, self.install_path))
                    os.chmod(self.install_path, 0o644)
            os.symlink(self.install_path, self.jar_link_path)
    
        def install(self):
            user = getpass.getuser()
            if user != 'root':
                log('only root can run this command: install')
                sys.exit(1)
    
            self.download_jar()
            extra_link_dirs = os.getenv("EXTRA_PATH", "")
            self.link_dirs.extend(extra_link_dirs.split(","))
            for d in self.link_dirs:
                self.ln_jar(d)
    
            # juicefs conf dir
            conf_dir = JUICEFS_CONF["juicefs.conf-dir"]
            if not os.path.exists(conf_dir):
                os.makedirs(conf_dir)
            conf_path = os.path.join(conf_dir, ".juicefs")
            os.system("rm -rf %s" % conf_path)
            conf_in_tmp = os.path.join(TEMP_DIR, ".juicefs")
            if os.path.exists(conf_in_tmp):
                if os.path.exists(conf_path):
                    shutil.rmtree(conf_path)
                shutil.move(conf_in_tmp, conf_dir)
            else:
                shutil.copytree("/root/.juicefs", conf_path)
            os.system('chmod 644 %s' % os.path.join(conf_path, '*'))
            log("copy JuiceFS conf to dir: %s" % conf_path)
    
            # juicefs cache dir
            if os.path.exists(self.core_site_path):
                content = open(self.core_site_path, "r").read()
                cache_dir = read_xml_conf(content, '.'.join(['juicefs', get_jfs_name(), 'cache-dir']))
                if not cache_dir:
                    cache_dir = read_xml_conf(content, 'juicefs.cache-dir')
                if cache_dir:
                    self.make_cache_dir(cache_dir)
    
            log("install JuiceFS on %s success" % socket.gethostname())
            return 0
    
        def _copy_to_host(self, items, host):
            # clean temp dir
            cmd = "%s %s \"sudo sh -c \'rm -rf %s;  mkdir %s; chmod 777 %s;\'\"" % (
                SSH, host, TEMP_DIR, TEMP_DIR, TEMP_DIR)
            code, txt = commands.getstatusoutput(cmd)
            if code > 0:
                log("\n%s, command is: %s" % (str(txt), cmd))
                return code >> 8
            for item in items:
                code = exec_cmd("scp -o StrictHostKeyChecking=no -r %s %s:%s" % (item, host, TEMP_DIR))
                if code > 0:
                    return code
            code = exec_cmd("%s %s sudo chmod 777 -R %s" % (SSH, host, os.path.join(TEMP_DIR, '*')))
            if code > 0:
                return code
            return 0
    
        def install_all(self):
            code, text = commands.getstatusoutput('sudo ls')
            if code > 0:
                log('only sudoer can run this script')
                sys.exit(1)
            script_path = sys.argv[0]
            local_resources = [self.install_path, script_path, get_conf_path()]
            jfs_name = get_jfs_name()
            extra_link_dirs = os.getenv("EXTRA_PATH", "")
            for host in self.discover_nodes():
                if in_localhost(host):
                    continue
    
                log("\ninstalling JuiceFS on %s" % host)
    
                # scp all needed to remote host
                code = self._copy_to_host(local_resources, host)
                if code > 0:
                    log("install JuiceFS on %s failed" % host)
                    continue
    
                # remote install
                code = os.system(
                    "%s %s JUICEFS_NAME=%s JFS_VERSION=%s DOWNLOAD_JUICEFS=False HADOOP_COMMAND=%s EXTRA_PATH=%s JAVA_HOME=%s sudo -E python %s install" % (
                        SSH, host, jfs_name, VERSION, HADOOP_CMD, extra_link_dirs, JAVE_HOME,
                        os.path.join(TEMP_DIR, os.path.basename(script_path))))
                if code > 0:
                    log("install JuiceFS on %s failed" % host)
    
        def uninstall(self):
            extra_link_dirs = os.getenv("EXTRA_PATH", "")
            self.link_dirs.extend(extra_link_dirs.split(","))
            for d in self.link_dirs:
                remove_jar(d)
            remove_jar(self.download_dir)
            log("uninstall JuiceFS on %s success" % socket.gethostname())
            return 0
    
        def uninstall_all(self):
            script_path = sys.argv[0]
            local_resources = [script_path]
            extra_link_dirs = os.getenv("EXTRA_PATH", "")
            for host in self.discover_nodes():
                log("\nuninstalling JuiceFS on %s" % host)
    
                # scp all needed to remote host
                code = self._copy_to_host(local_resources, host)
                if code > 0:
                    log("install JuiceFS on %s failed" % host)
                    continue
    
                # remote uninstall
                code = os.system(
                    "%s %s HADOOP_COMMAND=%s EXTRA_PATH=%s sudo -E python %s uninstall" % (
                        SSH, host, HADOOP_CMD, extra_link_dirs,
                        os.path.join(TEMP_DIR, os.path.basename(script_path))))
                if code > 0:
                    log("uninstall JuiceFS on %s failed" % host)
    
        def test(self):
            # self._get_input_param()
            jfs_name = get_jfs_name()
            success = True
    
            # test if conf has been deployed yet
            configured_conf = self.query_configured_conf()
            log('testing juicefs conf has been deployed')
            configured_core_site = dict(
                filter(lambda elem: self.decide_conf_file(elem[0]) == 'core-site', configured_conf.items()))
            configured_mapred_site = dict(
                filter(lambda elem: self.decide_conf_file(elem[0]) == 'mapred-site', configured_conf.items()))
    
            deployed_core_site = self.query_deployed_core_site()
            conf = self.diff_conf(deployed_core_site, configured_core_site)
    
            if len(conf.items()) > 0:
                log('ERROR: core-site.xml has not been deployed to your cluster yet')
                log(json.dumps(conf))
                success = False
    
            deployed_mapred_site = self.query_deployed_mapred_site()
            conf = self.diff_conf(deployed_mapred_site, configured_mapred_site)
    
            if len(conf.items()) > 0:
                log('ERROR: mapred-site.xml has not been deployed to your cluster yet')
                log(json.dumps(conf))
                success = False
            if success:
                log('juicefs conf has been deployed to cluster')
    
            # test if jar was installed in the right place
            log('testing if juicefs-hadoop.jar was installed in the right place')
            for path in self.link_dirs:
                if path is None or not os.path.exists(path):
                    continue
                if not os.path.exists(os.path.join(path, JAR_NAME)):
                    success = False
                    log("ERROR: %s was not found in %s" % (path, socket.gethostname()))
            path = os.path.join(get_conf_path(), '%s.conf' % jfs_name)
            if not os.path.exists(path):
                success = False
                log("ERROR: %s was not found in %s" % (path, socket.gethostname()))
            if success:
                log('juicefs-hadoop.jar has been placed in the right place')
    
            # test hadoop fs cmd
            log('testing hadoop fs shell cmd')
            cmds = [
                "%s fs -ls jfs://%s/ " % (HADOOP_CMD, jfs_name)
                , "%s fs -put /etc/hosts jfs://%s/juicefs-test-hosts " % (HADOOP_CMD, jfs_name)
                , "%s fs -cat jfs://%s/juicefs-test-hosts " % (HADOOP_CMD, jfs_name)
                , "%s fs -rm -skipTrash jfs://%s/juicefs-test-hosts " % (HADOOP_CMD, jfs_name)
            ]
            for cmd in cmds:
                code, text = commands.getstatusoutput(cmd)
                if code > 0:
                    success = False
                    # show the res
                    log(text)
    
            test_processes = ['org.apache.hadoop.hive.metastore.HiveMetaStore',
                              'org.apache.hive.service.server.HiveServer2',
                              'org.apache.hadoop.yarn.server.nodemanager.NodeManager',
                              'IMPALAD']
            if success:
                log('hadoop fs shell command success')
    
            # test process is restarted
            log('testing if related process has been restarted')
            for p in test_processes:
                code, text = commands.getstatusoutput('sudo %s -lvm | grep %s' % (os.path.join(JAVE_HOME, 'bin/jps'), p))
                if code > 0:
                    continue
                pid = text.split()[0]
                code, text = commands.getstatusoutput('sudo lsof -p %s | grep %s' % (pid, JAR_FULL_NAME))
                if code > 0:
                    success = False
                    log('ERROR: %s has not been restarted yet' % p)
            if success:
                log('all related process has been restarted')
    
            if success:
                log("test JuiceFS on %s success" % socket.gethostname())
                return 0
            else:
                log("test JuiceFS on %s failed" % socket.gethostname())
                return 1
    
        def query_deployed_core_site(self):
            if not os.path.exists(self.core_site_path):
                log("%s not existed on %s" % (self.core_site_path, socket.gethostname()))
                return dict()
            content = open(self.core_site_path, "r").read()
            result = list_all_xml_conf(content)
            return result
    
        def query_deployed_mapred_site(self):
            result = dict()
            if os.path.exists(self.mapred_site_path):
                content = open(self.mapred_site_path, "r").read()
                key = "mapreduce.application.classpath"
                value = read_xml_conf(content, key)
                if value:
                    result[key] = value
            return result
    
        def test_all(self):
            script_path = sys.argv[0]
            local_resources = [script_path]
            jfs_name = get_jfs_name()
            for host in self.discover_nodes():
                log("\ntesting JuiceFS on %s" % host)
    
                # scp all needed to remote host
                code = self._copy_to_host(local_resources, host)
                if code > 0:
                    log("test JuiceFS on %s failed" % host)
                    continue
    
                # remote test
                env_dic = {"JUICEFS_NAME": jfs_name,
                           "CONFIGURED_CONF": "\\''%s'\\'" % json.dumps(self.query_configured_conf())}
                code = self.ssh_exec_cmd(host, 'test', env_dic)
                if code > 0:
                    log("test JuiceFS on %s failed" % host)
    
        def config(self):
            self._get_input_param()
    
            configured_conf = self.query_configured_conf()
            self.current_conf = configured_conf
    
            jfs_name = get_jfs_name()
            default_core_site = self.parse_conf(os.path.join(get_conf_path(), jfs_name + ".conf"))
            merged_conf = configured_conf.copy()
            for k, v in default_core_site.items():
                if not has_jfs_conf(k, merged_conf):
                    self.merge_conf(merged_conf, k, v)
    
            cache_dir = merged_conf.get('juicefs.cache-dir', merged_conf.get("juicefs.%s.cache-dir" % jfs_name))
            cache_dir = raw_input("please input \"juicefs.%s.cache-dir\"(%s):" % (jfs_name, cache_dir)) or cache_dir
            cache_size = merged_conf.get('juicefs.cache-size', merged_conf.get("juicefs.%s.cache-size" % jfs_name))
            if '/dev/shm' in cache_dir:
                cache_size = '1024'
                merged_conf['juicefs.cache-replica'] = '0'
            cache_size = raw_input("please input \"juicefs.%s.cache-size\"(%s):" % (jfs_name, cache_size)) or cache_size
    
            self.merge_conf(merged_conf, 'juicefs.cache-dir', cache_dir)
            self.merge_conf(merged_conf, 'juicefs.cache-size', cache_size)
    
            self.update_conf(merged_conf)
            self._print_all_conf()
            # if len(self.need_to_add_conf.items()) > 0:
            #     self.deploy_config()
            self.make_cache_dir_all(cache_dir)
            log("config success")
    
        def make_cache_dir(self, cache_dir=None):
            user = getpass.getuser()
            if user != 'root':
                log('only root can run this command: make_cache_dir')
                sys.exit(1)
            if cache_dir is None:
                cache_dir = os.getenv('JUICEFS_CACHE_DIR', JUICEFS_CONF['juicefs.cache-dir'])
            patterns = cache_dir.split(':')
            for pattern in patterns:
                dirs = self._expand_dir(pattern)
                for dir in dirs:
                    if not os.path.exists(dir):
                        os.makedirs(dir)
                        os.chmod(dir, 0o777)
            log("make cache dir on %s success" % socket.gethostname())
    
        def _expand_dir(self, pattern):
            if pattern.endswith('/'):
                pattern = pattern[:len(pattern) - 1]
            if pattern == '':
                return ['/']
            matched = glob.glob(pattern)
            if len(matched) > 0:
                return matched
            dir, file = os.path.split(pattern)
            dirs = self._expand_dir(dir)
            return [os.path.join(x, file) for x in dirs]
    
        def make_cache_dir_all(self, cache_dir):
    
            self._get_input_param()
    
            script_path = sys.argv[0]
            local_resources = [script_path]
            hosts = self.discover_nodes()
            if socket.gethostname() not in hosts:
                hosts.extend([socket.gethostname()])
            for host in hosts:
                log('make cache dir on %s' % host)
                # scp all needed to remote host
                code = self._copy_to_host(local_resources, host)
                if code > 0:
                    log("make cache dir on %s failed" % host)
                    continue
    
                code = self.ssh_exec_cmd(host, "make_cache_dir", {"JUICEFS_CACHE_DIR": cache_dir})
                if code > 0:
                    log("make cache dir on %s failed" % host)
    
        def query_configured_conf(self):
            configured_conf = os.getenv('CONFIGURED_CONF')
            if configured_conf:
                return json.loads(configured_conf)
            self._get_input_param()
            configured_core_site = self.query_configured_core_site()
            configured_mapred_site = self.query_configured_mapred_site()
            configured_core_site.update(configured_mapred_site)
            return configured_core_site
    
        def query_configured_core_site(self):
            return self.query_deployed_core_site()
    
        def query_configured_mapred_site(self):
            return self.query_deployed_mapred_site()
    
        def update_conf(self, conf_dict):
            core_conf = dict(
                filter(lambda elem: self.decide_conf_file(elem[0]) == 'core-site', conf_dict.items()))
            mapred_conf = dict(
                filter(lambda elem: self.decide_conf_file(elem[0]) == 'mapred-site', conf_dict.items()))
            self.update_core_site(core_conf)
            self.update_mapred_site(mapred_conf)
            self.update_hive()
    
        def update_core_site(self, conf_dict):
            log(
                "Auto update core-site.xml is not supported in this Hadoop distribution, "
                "please configure it manually")
    
        def update_mapred_site(self, conf_dict):
            pass
    
        def update_hive(self):
            pass
    
        def diff_conf(self, get, want):
            diff_results = dict()
            for k, v in want.items():
                key = k.strip()
                if key not in get.keys():
                    diff_results[key] = v
                if key in get.keys():
                    if isinstance(v, list):
                        for i in v:
                            if i != get[key]:
                                diff_results[key] = i
                    else:
                        if v != get[key]:
                            diff_results[key] = v
            return diff_results
    
        def merge_conf(self, dic, k, v):
            self.need_to_add_conf[k] = v
            dic[k] = v
    
        def decide_conf_file(self, key):
            if key.startswith('juicefs') or key in ['fs.jfs.impl', 'fs.AbstractFileSystem.jfs.impl']:
                return 'core-site'
            elif key.strip() == 'mapreduce.application.classpath':
                return 'mapred-site'
            elif key.strip() == 'hive_aux_jars':
                return 'hive-interactive-env'
            elif key.strip() == 'ranger.plugin.hive.urlauth.filesystem.schemes':
                return 'ranger-hive-security'
            return 'core-site'
    
        def _print_all_conf(self):
            def print_conf(conf_dict):
                lst = dict(
                    map(lambda kv: ('='.join([kv[0], json.dumps(kv[1]).replace('"', '')]), self.decide_conf_file(kv[0])),
                        conf_dict.items())).items()
                from itertools import groupby
                # lst.sort(key=lambda x: x[1])
                lstg = groupby(lst, lambda x: x[1])
                for key, group in lstg:
                    log("%s: " % key)
                    for g in group:
                        log('\t%s' % g[0])
    
            log("\ncurrent configurations are:")
            print_conf(self.current_conf)
            if len(self.need_to_add_conf.items()) == 0:
                log('\nnothing was going to add to your cluster')
            else:
                log('\nthose configurations below will be added to cluster:')
                print_conf(self.need_to_add_conf)
    
        def deploy_config(self):
            log("deploy config is not supported in this Hadoop distribution")
    
        def discover_nodes(self):
            node_list = os.getenv("NODE_LIST")
            if node_list:
                return node_list.split(",")
            return self.do_discover_nodes()
    
        def do_discover_nodes(self):
            log("Can't find the nodes in cluster, you can define env variable NODE_LIST like this:")
            log("export NODE_LIST=node1,node2")
            log("python setup-hadoop.py COMMAND")
            sys.exit(1)
    
        def _get_input_param(self):
            self.username = "admin"
            self.password = "admin"
    
        def do_post(self, url, data='', header=None):
            return do_request(url, self.username, self.password, data=data, request_type='POST', header=header).getcode()
    
        def do_get(self, url, header=None):
            return do_request(url, self.username, self.password, request_type='GET', header=header).read()
    
        def do_put(self, url, data='', header=None):
            return do_request(url, self.username, self.password, data=data, request_type='PUT', header=header).getcode()
    
        def restart(self):
            log(
                "Auto restart is not supported in this Hadoop distribution, "
                "please restart it manually")
            return 0
    
        def ssh_exec_cmd(self, host, command, environment):
            script_path = sys.argv[0]
            local_resources = [script_path]
            code = self._copy_to_host(local_resources, host)
            if code > 0:
                return code
    
            environment["HADOOP_COMMAND"] = HADOOP_CMD
            env = ' '.join(['='.join([x[0], x[1]]) for x in environment.items()])
    
            cmd = "%s %s %s sudo -E python %s %s" % (
                SSH, host, env, os.path.join(TEMP_DIR, os.path.basename(script_path)), command)
            code = os.system(cmd)
            if code > 0:
                return code >> 8
    
        # def shrink_hdfs(self):
        #     self._get_input_param()
        #
        #     roll_interval = 60
        #
        #     def cp_data(hostname):
        #         code = self.ssh_exec_cmd(hostname, 'shrink_datanode_cp', {'DATANODE_DIR1': dir1, 'DATANODE_DIR2': dir2})
        #         if code > 0:
        #             log("shrink datanode on %s failed" % hostname)
        #         else:
        #             log("rsync data from %s to %s success" % (dir1, dir2))
        #
        #     grouped_datanodes = self.discover_datanodes()
        #     for group, datanodes, data_dir_list in grouped_datanodes:
        #         log("shrinking hdfs datanodes on this group: %s, %s" % (group, json.dumps(datanodes)))
        #         dirs = data_dir_list.split(',')
        #         dir1 = ''
        #         dir2 = ''
        #         while dir1 not in dirs:
        #             dir1 = raw_input("please input the disk dir you want remove(%s):" % data_dir_list)
        #         dir2 = raw_input("please input the disk dir you want move data to:")
        #         pool = ThreadPool(1)
        #         # first copy all data from one disk to the other disk
        #         for host in datanodes:
        #             pool.add_task(cp_data, host)
        #             log("copying data from %s to %s" % (dir1, dir2))
        #         pool.wait_completion()
        #         # then stop the datanode and rsync the rest to the other disk again and start the datanode
        #         # for host in datanodes:
        #         #     log("completing shrink_datanode on %s" % host)
        #         #     self.stop_datanode(host)
        #         #     cp_data(host)
        #         #     self.start_datanode(host)
        #         #     log("shrink datanode on %s success" % host)
        #         #     time.sleep(roll_interval)
    
        def query_datanode_dir(self, host):
            """
            :param host:
            :return: string
            """
            content = open(self.hdfs_site_path, "r").read()
            conf = list_all_xml_conf(content)
            if 'dfs.data.dir' in conf.keys():
                return conf['dfs.data.dir']
            else:
                return conf['dfs.datanode.data.dir']
    
        def _list_blk_dirs(self, dfs_data_dir, dirs):
            items = os.listdir(dfs_data_dir)
            for i in items:
                path = os.path.join(dfs_data_dir, i)
                if os.path.isfile(path) and i.startswith("blk_"):
                    dirs.append(dfs_data_dir)
                    break
                elif os.path.isdir(path):
                    self._list_blk_dirs(path, dirs)
    
        def _split_dir(self, dfs_data_dirs, split_num):
            dirs_list_files = []
    
            dir_to_list_files = {}
    
            # prepare dfs_data_dir split files
            for i in range(len(dfs_data_dirs)):
                list_files = []
                for j in range(split_num):
                    list_files.append("/tmp/shrink_datanode_rsync_src_%d_%d" % (i, j))
                dir_to_list_files[dfs_data_dirs[i]] = list_files
    
            # write splitted dfs_data_dir to split files
            for src_dir, split_files in dir_to_list_files.items():
                dirs = []
                self._list_blk_dirs(src_dir, dirs)
                if len(dirs) > 0:
                    fos = [open(f, 'w') for f in split_files]
                    for d in dirs:
                        index = abs(hash(d)) % split_num
                        fos[index].write('%s\n' % d)
                    for fo in fos:
                        fo.close()
                else:
                    log("no data in this datanode")
                    sys.exit(0)
    
            return dir_to_list_files
    
        def shrink_datanode(self):
            self._get_input_param()
            host = os.getenv('SLAVE_HOST', socket.gethostname())
            data_dir_list = self.query_datanode_dir(host)
            dfs_data_dirs = [os.path.abspath(x) for x in data_dir_list.split(",")]
            src = raw_input("please input the disk dir you want remove(%s):" % data_dir_list) or os.getenv('DATA_DIR_SRC')
            dst = raw_input("please input the disk dir you want move data to(%s):" % data_dir_list) or os.getenv(
                'DATA_DIR_DST')
            src = os.path.abspath(src)
            dst = os.path.abspath(dst)
            src_dirs = [os.path.abspath(x) for x in src.split(",")]
            dst_dirs = [os.path.abspath(x) for x in dst.split(",")]
            dst_dir_num = len(dst_dirs)
            src_dir_to_split_files = self._split_dir(src_dirs, dst_dir_num)
    
            self._check_dir(src_dirs, dst_dirs, dfs_data_dirs, src_dir_to_split_files)
            # first rsync
            for src_dir, file_list in src_dir_to_split_files.items():
                for i in range(dst_dir_num):
                    temp_dir = os.path.join(get_mount_point(dst_dirs[i]), 'shrink_temp')
                    rsync_data(src_dir, temp_dir, files_from=file_list[i])
            # stop datanode
            self.stop_datanode(host)
            # second rsync
            for src_dir, file_list in src_dir_to_split_files.items():
                for i in range(dst_dir_num):
                    temp_dir = os.path.join(get_mount_point(dst_dirs[i]), 'shrink_temp')
                    rsync_data(src_dir, temp_dir, files_from=file_list[i])
            # merge
            for src_dir, file_list in src_dir_to_split_files.items():
                for i in range(dst_dir_num):
                    temp_dir = os.path.join(get_mount_point(dst_dirs[i]), 'shrink_temp')
                    temp_dir += src_dir
                    rsync_data(temp_dir, dst_dirs[i], hard_link=True, ignore_existing=True, remove_source_files=True)
            # remove dfs_data_dir
            for src_dir in src_dirs:
                self.remove_datanode_dir(host, src_dir)
            # start datanode
            self.start_datanode(host)
            # clean
            for i in range(dst_dir_num):
                temp_dir = os.path.join(get_mount_point(dst_dirs[i]), 'shrink_temp')
                self._clean(temp_dir)
            # check hdfs
            self._check_hdfs()
            log('shrink datanode on %s success' % host)
    
        def _clean(self, temp_dir):
            log('cleaning temp dir: %s' % temp_dir)
            white_list = ['VERSION', 'dfsUsed', 'scanner.cursor']
            for root, dirs, files in os.walk(temp_dir):
                if len(files) > 0:
                    for name in files:
                        if name not in white_list:
                            log("file not merged: %s" % os.path.join(root, name))
                            sys.exit(1)
            shutil.rmtree(temp_dir)
    
        def _check_hdfs(self):
            cmd = 'sudo -u %s hdfs fsck /' % self.hdfs_superuser
            log("checking hdfs health use command: %s" % cmd)
            code = exec_cmd(cmd)
            if code > 0:
                log('hdfs check failed')
                sys.exit(1)
    
        def stop_datanode(self, host):
            log(
                "Auto stop datanode is not supported in this Hadoop distribution, "
                "please stop it manually")
    
        def start_datanode(self, host):
            log(
                "Auto start datanode is not supported in this Hadoop distribution, "
                "please start it manually")
    
        def _check_dir(self, src_dirs, dst_dirs, dfs_data_dirs, src_dir_to_split_files):
    
            src_dirs = [os.path.abspath(x) for x in src_dirs]
            dst_dirs = [os.path.abspath(x) for x in dst_dirs]
            dfs_data_dirs = [os.path.abspath(x) for x in dfs_data_dirs]
    
            # check dir exist on node and in data dir list
            def _check_dir_exist(dirs):
                for d in dirs:
                    if not os.path.exists(d):
                        log("%s not exist on %s" % (d, socket.gethostname()))
                        sys.exit(1)
                    if d not in dfs_data_dirs:
                        log("%s not in datanode data dirs: %s" % (d, ','.join(dfs_data_dirs)))
                        sys.exit(1)
            _check_dir_exist(src_dirs)
            _check_dir_exist(dst_dirs)
    
            # check if process datanode exsists
            sts, text = commands.getstatusoutput('sudo %s -lvm | grep %s' % (
                os.path.join(JAVE_HOME, 'bin/jps'), 'org.apache.hadoop.hdfs.server.datanode.DataNode'))
            if sts > 0:
                log('%s do not have datanode run on it' % socket.gethostname())
                sys.exit(1)
    
            # check if the provided dir are on the same device
            for src_dir in src_dirs:
                for dst_dir in dst_dirs:
                    if get_mounted_device(src_dir) == get_mounted_device(dst_dir):
                        log('%s and %s are on the same disk' % (src_dir, dst_dir))
                        sys.exit(1)
    
            # check space
            src_dirs_space = 0
            for src_dir in src_dirs:
                dir_size = get_dir_size(src_dir)
                log('to remove dir: %s, size: %d G' % (src_dir, dir_size >> 30))
                src_dirs_space += dir_size
            dst_dirs_space = 0
            for dst_dir in dst_dirs:
                dir_size = get_fs_freespace(dst_dir)
                log('to move data disk %s, size: %d G' % (get_mount_point(dst_dir), dir_size >> 30))
                dst_dirs_space += dir_size
            ratio = 1.2
            if src_dirs_space * ratio > dst_dirs_space:
                for dst_dir in dst_dirs:
                    mount_point = get_mount_point(dst_dir)
                    log('%s may not have enough room for data' % mount_point)
                sys.exit(1)
    
        def discover_datanodes(self):
            return {}
    
        def remove_datanode_dir(self, host, dir):
            log(
                "Auto discover yarn resourcemanagers is not supported in this Hadoop distribution, "
                "please configure it manually")
    
        def benchmark(self, args=None, results=None):
            """
            python setup-hadoop.py benchmark fsio
            :return:
            """
            import argparse
            default_test_dir = '/benchmarks'
            default_file_size = '5000'  # mb
            default_file_num = '50000'  # number of files
    
            parser = argparse.ArgumentParser(prog="python setup-hadoop.py benchmark TYPE", description='benchmark JuiceFS')
            parser.add_argument("method", help="benchmark")
    
            subparsers = parser.add_subparsers(help='benchmark type')
            parser_all = subparsers.add_parser('all', help='benchmark all')
            parser_fsio = subparsers.add_parser('fsio', help='i/o benchmark')
            parser_dfsio = subparsers.add_parser('dfsio', help='Distributed i/o benchmark')
            parser_nnbench = subparsers.add_parser('nnbench', help='A benchmark that stresses the namenode')
            parser_nnbench_local = subparsers.add_parser('nnbench_local',
                                                         help='A benchmark that stresses the namenode in local process')
            # all benchmark
            parser_all.set_defaults(func=self._benchmark_all)
            parser_all.add_argument("-testDir", help='which fs to run benchmark', default=default_test_dir)
            parser_all.add_argument("-fileSize", help='size in MB, used in io benchmark', default='5000')
            parser_all.add_argument("-numFiles", help='number of files, used in meta benchmark', default='50000')
    
            # fsio benchmark
            group = parser_fsio.add_mutually_exclusive_group()
            group.add_argument("-read", action="store_true", help='benchmark read')
            group.add_argument("-write", action="store_true", help='benchmark write')
            parser_fsio.add_argument("-random", action="store_true", help='benchmark random read')
            parser_fsio.add_argument("-fileSize", help='size in MB', default=default_file_size)
            parser_fsio.add_argument("-nrFiles", help='number of maps', default='1')
            parser_fsio.add_argument("-bufferSize", help='size in Bytes', default='524288')
            parser_fsio.add_argument("-threadsPerMap", help='number of threads in a single process', default='1')
            parser_fsio.add_argument("-baseDir", help='base dir', default='%s/%s' % (default_test_dir, 'fsio'))
            parser_fsio.set_defaults(func=self._benchmark_fsio)
    
            # dfsio benchmark
            group = parser_dfsio.add_mutually_exclusive_group()
            group.add_argument("-read", action="store_true", help='benchmark read')
            group.add_argument("-write", action="store_true", help='benchmark write')
            parser_dfsio.add_argument("-random", action="store_true", help='benchmark random read')
            parser_dfsio.add_argument("-fileSize", help='size in MB', default=default_file_size)
            parser_dfsio.add_argument("-nrFiles", help='number of maps', default='1')
            parser_dfsio.add_argument("-bufferSize", help='size in Bytes', default='524288')
            parser_dfsio.add_argument("-threadsPerMap", help='number of threads in a single process', default='1')
            parser_dfsio.add_argument("-baseDir", help='base dir', default='%s/%s' % (default_test_dir, 'dfsio'))
            parser_dfsio.set_defaults(func=self._benchmark_dfsio)
    
            # nnbench_local
            parser_nnbench_local.add_argument("-operation", help='createWrite openRead rename delete',
                                              choices=['createWrite', 'openRead', 'rename', 'delete'])
            parser_nnbench_local.add_argument("-numFiles", help='number of files', default=default_file_num)
            parser_nnbench_local.add_argument("-bytesPerBlock", help='blocksize', default='134217728')
            parser_nnbench_local.add_argument("-baseDir", help='base dir',
                                              default='%s/%s' % (default_test_dir, 'nnbench_local'))
            parser_nnbench_local.set_defaults(func=self._benchmark_nnbench_local)
    
            # nnbench
            parser_nnbench.add_argument("-operation", help='create_write open_read rename delete',
                                        choices=['create_write', 'open_read', 'rename', 'delete'])
            parser_nnbench.add_argument("-maps", help='number of maps', default='1')
            parser_nnbench.add_argument("-numberOfFiles", help='number of files', default=default_file_num)
            parser_nnbench.add_argument("-blockSize", help='blocksize', default='134217728')
            parser_nnbench.add_argument("-threadsPerMap", help='number of threads in a single map', default='1')
            parser_nnbench.add_argument("-baseDir", help='base dir', default='%s/%s' % (default_test_dir, 'nnbench'))
            parser_nnbench.set_defaults(func=self._benchmark_nnbench)
    
            args = parser.parse_args(args)
            result = args.func(args)
            if results is not None:
                results.append(result)
    
        def _benchmark_fsio(self, args):
            os.system("sync; echo 3 > /proc/sys/vm/drop_caches")
            command = JAVE_HOME + '/bin/java -Xmx200m -cp `%s classpath`:/usr/local/lib/juicefs-hadoop.jar com.juicefs.bench.TestFSIO' % HADOOP_CMD
            if args.read:
                command += " -read"
                if args.random:
                    command += " -random"
            elif args.write:
                command += " -write"
            command += " -fileSize %s" % args.fileSize
            command += " -bufferSize %s" % args.bufferSize
            command += " -threadsPerMap %s" % args.threadsPerMap
            command += " -baseDir %s" % args.baseDir
            log('executing: %s' % command)
            import subprocess
            p = subprocess.Popen(command, shell=True, stderr=subprocess.PIPE)
            cpu, mem, recv, sent = self._get_statis(p)
            out, err = p.communicate()
            text = "\n".join(err.splitlines())
            code = p.poll()
            if code > 0:
                sys.exit(1)
            else:
                speed = extract_number(text, 'Total Throughput mb/sec: ')
                time = extract_number(text, 'Test exec time sec: ')
                if args.read:
                    op = 'read'
                else:
                    op = 'write'
                res = '%s %smb, speed: %smb/s; cpu: %.1f%%, mem: %dmb, recv: %.1fmb/s, sent:  %.1fmb/s' \
                      % (op, args.fileSize, speed, cpu, mem, recv, sent)
                return res
    
        def _get_statis(self, process):
            try:
                import psutil
            except Exception:
                log('please run pip install psutil if you want to get cpu, mem, recv, send')
                return 0, 0, 0, 0
            time.sleep(5)  # skip first 5s
            p = psutil.Process(process.pid)
            cnt = 0
            percent = 0.0
            memory = 0
            o_recv = psutil.net_io_counters().bytes_recv
            l_recv = 0
            o_sent = psutil.net_io_counters().bytes_sent
            l_sent = 0
            while True:
                try:
                    if p.status() == "zombie":
                        break
                    percent += p.cpu_percent(interval=1)
                    memory += (p.memory_full_info().rss >> 20)
                    l_recv = psutil.net_io_counters().bytes_recv
                    l_sent = psutil.net_io_counters().bytes_sent
                    cnt += 1
                except Exception:
                    break
            # print ("cpu: %.1f%%, mem: %dmb, recv: %.1fmb/s, sent:  %.1fmb/s" % (
            #     percent, memory, (((l_recv - o_recv) >> 20) / float(cnt)), (((l_sent - o_sent) >> 20) / float(cnt))))
            return percent / float(cnt), \
                   memory / float(cnt), \
                   (((l_recv - o_recv) >> 20) / float(cnt)), \
                   (((l_sent - o_sent) >> 20) / float(cnt))
    
        @staticmethod
        def _benchmark_dfsio(args):
            command = "%s jar /usr/local/lib/juicefs-hadoop.jar com.juicefs.bench.TestDFSIO" % HADOOP_CMD
            if args.read:
                command += " -read"
                if args.random:
                    command += " -random"
            elif args.write:
                command += " -write"
            command += " -fileSize %s" % args.fileSize
            command += " -bufferSize %s" % args.bufferSize
            command += " -nrFiles %s" % args.nrFiles
            command += " -threadsPerMap %s" % args.threadsPerMap
            command += " -baseDir %s" % args.baseDir
            log('executing: %s' % command)
            sts, text = commands.getstatusoutput(command)
            if sts > 0:
                log(text)
                sys.exit(sts >> 8)
            else:
                speed = extract_number(text, 'Total Throughput mb/sec: ')
                time = extract_number(text, 'Test exec time sec: ')
                if args.read:
                    op = 'read'
                else:
                    op = 'write'
                res = '%s %smb in %s seconds, total speed: %smb/sec' % (op, args.fileSize, time, speed)
                return res
    
        @staticmethod
        def _benchmark_nnbench_local(args):
            command = "%s jar /usr/local/lib/juicefs-hadoop.jar com.juicefs.bench.NNBenchWithoutMR" % HADOOP_CMD
            command += " -operation %s" % args.operation
            command += " -numFiles %s" % args.numFiles
            command += " -bytesPerBlock %s" % args.bytesPerBlock
            command += " -baseDir %s" % args.baseDir
            log('executing: %s' % command)
            sts, text = commands.getstatusoutput(command)
            if sts > 0:
                log(text)
                sys.exit(sts >> 8)
            else:
                ss = 'The %s TPS ' % args.operation
                tps = extract_number(text, ss)
                if 'createWrite' == args.operation:
                    tps = str(int(tps) / 2)
                ss = 'The %s job took ' % args.operation
                time = extract_number(text, ss)
                res = '%s %s files in %s seconds, TPS: %s' % (args.operation, args.numFiles, time, tps)
                return res
    
        @staticmethod
        def _benchmark_nnbench(args):
            command = "%s jar /usr/local/lib/juicefs-hadoop.jar com.juicefs.bench.NNBench" % HADOOP_CMD
            command += " -operation %s" % args.operation
            command += " -maps %s" % args.maps
            command += " -threadsPerMap %s" % args.threadsPerMap
            command += " -numberOfFiles %s" % args.numberOfFiles
            command += " -blockSize %s" % args.blockSize
            command += " -baseDir %s" % args.baseDir
            log('executing: %s' % command)
            sts, text = commands.getstatusoutput(command)
            if sts > 0:
                log(text)
                sys.exit(sts >> 8)
            else:
                if args.operation == 'create_write':
                    ss = 'TPS: Create/Write/Close: '
                elif args.operation == 'open_read':
                    ss = 'TPS: Open/Read: '
                elif args.operation == 'rename':
                    ss = 'TPS: Rename: '
                else:
                    ss = 'TPS: Delete: '
                tps = extract_number(text, ss)
                if 'create_write' == args.operation:
                    tps = str(int(tps) / 2)
                ss = 'Longest Map Time $ms$: '
                time = extract_number(text, ss)
                res = '%s %s files in %s seconds, TPS: %s' % (args.operation,
                                                              str(int(args.numberOfFiles) * int(args.maps) * int(
                                                                  args.threadsPerMap)),
                                                              str(float(time) / 1000), tps)
                return res
    
        def _benchmark_all(self, args):
            test_dir = args.testDir
            file_size = args.fileSize
            file_num = args.numFiles
    
            command_lines = ['fsio -write -fileSize %s -baseDir %s' % (file_size, test_dir + '/fsio'),
                             'fsio -read -fileSize %s -baseDir %s' % (file_size, test_dir + '/fsio'),
                             'nnbench_local -operation createWrite -numFiles %s -baseDir %s' % (
                                 file_num, test_dir + '/nnbench_local'),
                             'nnbench_local -operation openRead -numFiles %s -baseDir %s' % (
                                 file_num, test_dir + '/nnbench_local'),
                             'nnbench_local -operation rename -numFiles %s -baseDir %s' % (
                                 file_num, test_dir + '/nnbench_local'),
                             'nnbench_local -operation delete -numFiles %s -baseDir %s' % (
                                 file_num, test_dir + '/nnbench_local')
                             ]
            results = []
            for line in command_lines:
                self.benchmark(['benchmark'] + line.split(), results)
            for res in results:
                log(res)
    
        def _build_hive_jfs_args(self):
            import argparse
            parser = argparse.ArgumentParser(prog="python setup-hadoop.py hive_meta",
                                             description='transfer data in hive to JuiceFS')
    
            hive_data = parser.add_argument_group("hive data")
            hive_data.add_argument("--dir", help="hive data dir")
            hive_data.add_argument("-m", "--maps", type=int, help="distcp maps", default=0)
    
            hive_meta = parser.add_argument_group("hive meta")
            hive_meta.add_argument("--schema", help="target schema")
            hive_meta.add_argument("--database", help="hive database")
            hive_meta.add_argument("--table", help="hive table")
            hive_meta.add_argument("--start_pt", help="hive table start partition")
            hive_meta.add_argument("--end_pt", help="hive table end partition")
    
            hive_check_engine = parser.add_argument_group("hive check")
            hive_check_engine.add_argument("--engine", help="hive check engine", default="hive -S -e ")
            return parser.parse_args(args=sys.argv[2:])
    
        def hive_data(self, log_file=None):
            log('distcping data')
            args = self._build_hive_jfs_args()
            options = "-update -pugpt "
            if args.maps:
                options += "-m %d " % args.maps
            src_dir = args.dir if args.dir.endswith("/") else args.dir + "/"
            if args.schema:
                dst_dir = args.schema + urlparse(src_dir).path
            else:
                dst_dir = "jfs://%s" % get_jfs_name() + urlparse(src_dir).path
            cmd = "hadoop distcp %s %s %s" % (options, src_dir, dst_dir)
            if log_file:
                cmd = "%s >> %s 2>&1" % (cmd, log_file)
            if os.system(cmd) > 0:
                sys.exit(1)
    
        def hive_meta(self, log_file=None):
            log('altering hive table location')
            args = self._build_hive_jfs_args()
            database = args.database
            table = args.table
            start_pt = args.start_pt
            end_pt = args.end_pt
            jfs_name = get_jfs_name()
            sts, table_loc_str = commands.getstatusoutput(
                "hive -S -e 'show create table %s.%s' | grep -A 1 -i location | grep -i -v location | grep -v 'Java HotSpot(TM)'" % (
                    database, table))
            if sts > 0:
                log("%s.%s does not exist or hive not working" % (database, table))
                sys.exit(1)
            sts, text = commands.getstatusoutput("hive -e 'show partitions %s.%s'" % (database, table))
            if sts > 0:
                log("%s.%s is not partitioned" % (database, table))
                # not partitioned table
                src_loc = table_loc_str.split('\n').pop()
                p = urlparse(src_loc.replace("'", "").strip()).path
                if args.schema:
                    jfs_path = args.schema + p.replace("\n", "")
                else:
                    jfs_path = "jfs://" + jfs_name + p.replace("\n", "")
                os.system("hive -e \"use %s; alter table %s set location '%s'\"" % (database, table, jfs_path))
            else:
                log("%s.%s is partitioned" % (database, table))
                # partitioned table
                pts = [x for x in text.split('\n') if ' ' not in x and 'OK' != x and '' != x]
    
                si = self._search_pt(lst=pts, v=start_pt)
                ei = self._search_pt(lst=pts, v=end_pt)
                if si > 0 and ei > 0:
                    pts = pts[si:ei + 1]
                elif si > 0:
                    pts = pts[si:]
                elif ei > 0:
                    pts = pts[:ei + 1]
    
                # build  generate locations sql
                build_location_sql_path = "/tmp/hive2jfs.locations.sql.%s.%s" % (database, table)
                location_sql = open(build_location_sql_path, "w")
    
                pt_strs = [",".join(["=".join([x.split("=")[0], "'%s'" % x.split("=")[1]]) for x in pt.split("/")]) for pt
                           in
                           pts]
                for ps in pt_strs:
                    line = "desc formatted %s.%s partition(%s);\n" % (database, table, ps)
                    location_sql.write(line)
                location_sql.close()
    
                # generate locations
                builded_location_path = "/tmp/hive2jfs.locations.%s.%s" % (database, table)
                sts, text = commands.getstatusoutput(
                    "hive -f %s | grep -i location | awk '{ print $2 }' > %s" % (
                        build_location_sql_path, builded_location_path))
    
                if sts > 0:
                    print(text)
                    sys.exit(1)
                lf = open(builded_location_path)
    
                # generate alter sql
                i = 0
                alter_sql_path = "/tmp/hive2jfs.alter.sql.%s.%s" % (database, table)
                alter_sql = open(alter_sql_path, "w")
                alter_sql.write("use %s;\n" % database)
                for loc in lf.readlines():
                    p = urlparse(loc).path
                    if args.schema:
                        jfs_path = args.schema + p.replace("\n", "")
                    else:
                        jfs_path = "jfs://" + jfs_name + p.replace("\n", "")
                    alter_sql.write("alter table %s partition(%s) set location '%s';\n" % (table, pt_strs[i], jfs_path))
                    i += 1
                alter_sql.close()
    
                # exec alter command
                cmd = "hive -f %s" % alter_sql_path
                if log_file:
                    cmd = "%s >> %s 2>&1" % (cmd, log_file)
                os.system(cmd)
    
        def _search_pt(self, lst, v):
            if not v or not lst:
                return -1
            low, high = 0, len(lst) - 1
            while low <= high:
                mid = low + ((high - low) >> 1)
                if lst[mid] < v:
                    low = mid + 1
                elif lst[mid] > v:
                    high = mid - 1
                else:
                    if lst[mid] == v:
                        return mid
                    elif mid != len(lst) - 1 and lst[mid + 1] > v:
                        return mid + 1
                    else:
                        low = mid + 1
            return -1
    
        def hive_jfs(self):
            args = self._build_hive_jfs_args()
            # args
            table = args.table
            database = args.database
            start_pt = args.start_pt
            ent_pt = args.end_pt
            engine = args.engine
    
            log_file = "/tmp/hive2jfs.log.%s.%s" % (database, table)
            log("logging to %s" % log_file)
    
            self.hive_data(log_file)
    
            log("counting old table")
            start_cond = ""
            if start_pt:
                start_cond = " and " + " and ".join(
                    [">=".join([x.split("=")[0], "'%s'" % x.split("=")[1]]) for x in start_pt.split("/")])
            end_cond = ""
            if ent_pt:
                end_cond = " and " + " and ".join(
                    ["<=".join([x.split("=")[0], "'%s'" % x.split("=")[1]]) for x in ent_pt.split("/")])
            sql = "select count(*) from %s.%s where 1=1 %s %s" % (database, table, start_cond, end_cond)
            sts, before = commands.getstatusoutput("%s '%s'" % (engine, sql))
            if sts > 0:
                log(before)
    
            self.hive_meta(log_file)
    
            log("counting new table")
            sts, after = commands.getstatusoutput("%s '%s'" % (engine, sql))
            if sts > 0:
                log(after)
    
            if before != after:
                log("before:\n%s" % before)
                log("after:\n%s" % after)
                sys.exit(1)
    
            if args.schema:
                schema = args.schema
            else:
                schema = "jfs://" + get_jfs_name()
            log("moving data in %s.%s [%s - %s] to %s successfully" % (database, table, start_pt, ent_pt, schema))
    
    
    class CDH(HadoopDist):
        def __init__(self):
            super(CDH, self).__init__()
            self.home = self._find_home()
            # self.download_dir = os.path.join(self.home, "jars")
            self.link_dirs = [os.path.join(self.home, app) for app in
                              ["lib/hadoop/lib", "spark/jars", "/var/lib/impala/", "lib/solr/webapps/solr/WEB-INF/lib"]] + [
                                 '/usr/lib/hadoop/lib']
            self.addr = None
            self.hdfs_name = None
            self.yarn_name = None
            self.host_rolename_dic = None
            self.in_sa_cluster = False
            if os.path.exists('/home/sa_cluster'):
                self.in_sa_cluster = True
                JUICEFS_CONF['juicefs.file.checksum'] = 'true'
    
        def _get_input_param(self):
            if self.addr is not None:
                return
            address = self._find_cloudera_manager_addr()
            address = raw_input("please input Cloudera Manager address(%s):" % address) or address
            hostname_port = address.split(":")
            while len(hostname_port) != 2:
                log("input wrong Cloudera Manager address, please input again")
                address = raw_input("please input Cloudera Manager address(%s):" % address) or address
                hostname_port = address.split(":")
            self.host = hostname_port[0].strip()
            self.port = hostname_port[1].strip()
            self.rest_api_url_base = "http://%s:%s/api/v11/" % (self.host, self.port)
            username = 'admin'
            password = 'admin'
            if self.in_sa_cluster:
                user = getpass.getuser()
                if user != 'sa_cluster':
                    log('this script should run by user \"sa_cluster\"')
                    sys.exit(1)
                command = 'spadmin config get client -m cloudera'
                sts, text = commands.getstatusoutput(command)
                if sts > 0:
                    log('%s failed' % command)
                else:
                    text = '\n'.join([line for line in text.split("\n") if 'cloudera client conf' not in line])
                    dic = json.loads(text)
                    username = dic['cloudera_manager_username']
                    password = dic['cloudera_manager_password']
            self.username = raw_input("please input Cloudera Manager admin user(%s):" % username) or username
            self.password = raw_input("please input Cloudera Manager admin password(%s):" % password) or password
            check_input(self.username, self.password, urljoin(self.rest_api_url_base, "clusters"))
            self.addr = "%s:%s@%s:%s" % (self.username, self.password, self.host, self.port)
            self.cluster_name = self._query_cluster_name()
    
        def deploy_config(self):
            self._get_input_param()
            url = urljoin(self.rest_api_url_base, "clusters/%s/commands/deployClientConfig" % self.cluster_name)
            assert self.do_post(url) == 200
            log("\nSuccessfully deployed all client configurations")
    
        def _query_conf_by_key(self, url, key):
            self._get_input_param()
            response = self.do_get(url)
            items = json.loads(response)["items"]
            values = list(filter(lambda x: x["name"] == key, items))
            if len(values) == 0:
                result = ''
            else:
                item = values[0]
                result = item.get('value')
                if result is None:
                    result = item['default']
            return result
    
        def _query_role_conf(self, cluster_name, service_name, role_name, key):
            """
            :return: string
            """
            url = urljoin(self.rest_api_url_base,
                          "clusters/%s/services/%s/roles/%s/config" % (cluster_name, service_name, role_name))
            return self._query_conf_by_key(url, key)
    
        def _query_role_group_conf(self, cluster_name, service_name, role_group, key):
            """
            :return: string
            """
            url = urljoin(self.rest_api_url_base,
                          "clusters/%s/services/%s/roleConfigGroups/%s/config?view=full" % (
                              cluster_name, service_name, role_group))
            return self._query_conf_by_key(url, key)
    
        def _query_service_conf(self, cluster_name, service_name, key):
            """
            :return: string
            """
            url = urljoin(self.rest_api_url_base, "clusters/%s/services/%s/config" % (cluster_name, service_name))
            return self._query_conf_by_key(url, key)
    
        def _query_role_groups(self, cluster_name, service_name, role_type):
            """
            :return: list(string)
            """
            url = urljoin(self.rest_api_url_base,
                          "clusters/%s/services/%s/roleConfigGroups" % (cluster_name, service_name))
            response = self.do_get(url)
            items = list(filter(lambda x: x["roleType"] == role_type, json.loads(response)["items"]))
            return [x["name"] for x in items]
    
        def _query_cluster_name(self):
            url = urljoin(self.rest_api_url_base, "clusters")
            response = self.do_get(url)
            clusters = json.loads(response)["items"]
            names = [x["displayName"] for x in clusters]
            if len(clusters) == 1:
                cluster_name = names[0]
            elif len(clusters) > 1:
                cluster_name = raw_input("please specify a cluster(%s):" % ", ".join(names)) or names[0]
            else:
                log("cluster has not been created yet")
                sys.exit(1)
            return cluster_name.replace(" ", '%20')
    
        def _query_service_name(self, cluster_name, service_type):
            url = urljoin(self.rest_api_url_base, "clusters/%s/services" % cluster_name)
            response = self.do_get(url)
            services = json.loads(response)["items"]
            names = list(filter(lambda x: x["type"].strip() == service_type, services))
            if len(names) == 1:
                service_name = names[0]["name"]
            else:
                log("%s has not been installed yet" % service_type)
                sys.exit(1)
            return service_name
    
        def _query_hdfs_name(self):
            if self.hdfs_name is not None:
                return self.hdfs_name
            self.hdfs_name = self._query_service_name(self.cluster_name, 'HDFS')
            return self.hdfs_name
    
        def _query_yarn_name(self):
            if self.yarn_name is not None:
                return self.yarn_name
            self.yarn_name = self._query_service_name(self.cluster_name, 'YARN')
            return self.yarn_name
    
        def _query_role_name(self, role, host_name):
            """
            :return: string
            """
            url = urljoin(self.rest_api_url_base,
                          "hosts/%s" % host_name)
            response = self.do_get(url)
            role_refs = json.loads(response)["roleRefs"]
            return list(filter(lambda x: role in x["roleName"], role_refs))[0]["roleName"]
    
        def _query_role_group_nodes(self, cluster_name, service_name, role_group):
            """
            :return:dict list(string)
            """
            url = urljoin(self.rest_api_url_base,
                          "clusters/%s/services/%s/roleConfigGroups/%s/roles" % (
                              cluster_name, service_name, role_group))
            return [self._query_hostname(x["hostRef"]["hostId"]) for x in json.loads(self.do_get(url))["items"]]
    
        def _query_hostname(self, host_id):
            u = urljoin(self.rest_api_url_base, "hosts/%s" % host_id)
            res = self.do_get(u)
            return json.loads(res)["hostname"]
    
        def _update_role_conf(self, cluster_name, service_name, role_name, key, value):
            url = urljoin(self.rest_api_url_base,
                          "clusters/%s/services/%s/roles/%s/config" % (cluster_name, service_name, role_name))
            data = {"items": [{"name": key, "value": value}]}
            code = self.do_put(url, json.dumps(data),
                               {"Content-Type": "application/json", "Accept": "application/json"})
            assert code == 200
    
        def _update_role_group_conf(self, cluster_name, service_name, role_group, key, value):
            url = urljoin(self.rest_api_url_base,
                          "clusters/%s/services/%s/roleConfigGroups/%s/config?view=full" % (
                              cluster_name, service_name, role_group))
            data = {"items": [{"name": key, "value": value}]}
            code = self.do_put(url, json.dumps(data),
                               {"Content-Type": "application/json", "Accept": "application/json"})
            assert code == 200
    
        def _update_service_conf(self, cluster_name, service_name, key, value):
            url = urljoin(self.rest_api_url_base, "clusters/%s/services/%s/config" % (cluster_name, service_name))
            data = {"items": [{"name": key, "value": value}]}
            code = self.do_put(url, json.dumps(data), {"Content-Type": "application/json", "Accept": "application/json"})
            assert code == 200
    
        def query_configured_core_site(self):
            self._get_input_param()
            hdfs_name = self._query_hdfs_name()
            value = self._query_service_conf(self.cluster_name, hdfs_name, 'core_site_safety_valve')
            content = ("<properties>" + value + "</properties>").encode()
            return list_all_xml_conf(content)
    
        def update_core_site(self, conf_dict):
            hdfs_name = self._query_hdfs_name()
            self._update_service_conf(self.cluster_name, hdfs_name, "core_site_safety_valve", self._dict_to_xml(conf_dict))
    
        def _dict_to_xml(self, dic):
            properties = ET.fromstring("<properties></properties>".encode())
            for k, v in dic.items():
                prop = ET.Element('property')
                name = ET.SubElement(prop, 'name')
                value = ET.SubElement(prop, 'value')
                name.text = k
                value.text = v
                properties.append(prop)
            return ET.tostring(properties, encoding="utf-8").replace('<properties>', "").replace("</properties>", "")
    
        def query_configured_mapred_site(self):
            self._get_input_param()
            yarn_name = self._query_yarn_name()
            yarn_gateway_groups = self._query_role_groups(self.cluster_name, yarn_name, 'GATEWAY')
            result = dict()
    
            for yg in yarn_gateway_groups:
                value = self._query_role_group_conf(self.cluster_name, yarn_name, yg, 'mapreduce_application_classpath')
                values = result.get('mapreduce.application.classpath', [])
                values.append(value)
                result['mapreduce.application.classpath'] = values
            return result
    
        def update_mapred_site(self, dic):
            yarn_name = self._query_yarn_name()
            yarn_gateway_groups = self._query_role_groups(self.cluster_name, yarn_name, 'GATEWAY')
            for group in yarn_gateway_groups:
                value = self._query_role_group_conf(self.cluster_name, yarn_name, group, 'mapreduce_application_classpath')
                if 'juicefs-hadoop' not in value:
                    paths = value.split(',')
                    paths.append(self.jar_link_path)
                    new_value = ','.join(paths)
                    self._update_role_group_conf(self.cluster_name, yarn_name, group, 'mapreduce_application_classpath',
                                                 new_value)
                    values = self.need_to_add_conf.get('mapreduce.application.classpath', [])
                    values.append(new_value)
                    self.need_to_add_conf['mapreduce.application.classpath'] = values
    
        def do_discover_nodes(self):
            self._get_input_param()
            url = urljoin(self.rest_api_url_base, "hosts")
            response = self.do_get(url)
            items = json.loads(response)["items"]
            nodes = [x["ipAddress"] for x in items]
            return nodes
    
        def install(self):
            super(CDH, self).install()
            self._install_spark()
    
        def _find_home(self):
            cf = configparser.ConfigParser()
            cf.read("/etc/cloudera-scm-agent/config.ini")
            try:
                return os.path.join(cf.get("General", "parcel_dir"), "CDH")
            except configparser.NoOptionError:
                return "/opt/cloudera/parcels/CDH/"
    
        def _find_cloudera_manager_addr(self):
            cf = configparser.ConfigParser()
            cf.read("/etc/cloudera-scm-agent/config.ini")
            try:
                host = cf.get("General", "server_host").strip()
                port = '7180'
                return ':'.join([host, port])
            except configparser.NoOptionError:
                return ':'.join([socket.gethostname(), '7180'])
    
        def _install_spark(self):
            spark_classpath_txt = "/etc/spark/conf/classpath.txt"
            spark2_classpath_txt = "/etc/spark2/conf/classpath.txt"
            if os.path.exists(spark_classpath_txt):
                self._add_jar_path(spark_classpath_txt)
            if os.path.exists(spark2_classpath_txt):
                self._add_jar_path(spark2_classpath_txt)
    
        def _add_jar_path(self, file_name):
            f2 = open("%s.bak" % file_name, "w")
            text = open(file_name).read()
            text = '\n'.join(line for line in text.split("\n") if 'juicefs-hadoop' not in line)
            if not text.endswith('\n'):
                text += '\n'
            text += self.jar_link_path
    
            f2.write(text)
            f2.close()
            os.rename("%s.bak" % file_name, file_name)
    
        def restart(self):
            self._get_input_param()
            url = urljoin(self.rest_api_url_base, "clusters/%s/commands/restart" % self.cluster_name)
            header = {'Content-Type': 'application/json', 'Accept': 'application/json'}
            data = json.dumps({'restartOnlyStaleServices': True, 'redeployClientConfiguration': True})
            code = self.do_post(url, data, header)
            assert code == 200
            # self.deploy_config()
    
        def _role_command(self, cluster_name, service_name, role_names, command):
            url = urljoin(self.rest_api_url_base,
                          'clusters/%s/services/%s/roleCommands/%s' % (cluster_name, service_name, command))
            header = {'Content-Type': 'application/json', 'Accept': 'application/json'}
            data = json.dumps(role_names)
            code = self.do_post(url, data, header)
            assert code == 200
    
        def _role_status(self, cluster_name, service_name, role_name):
            url = urljoin(self.rest_api_url_base,
                          'clusters/%s/services/%s/roles/%s' % (cluster_name, service_name, role_name))
            response = self.do_get(url)
            return json.loads(response)['roleState']
    
        def start_datanode(self, host):
            self._get_input_param()
            log('starting datanode on %s' % host)
            hdfs_name = self._query_hdfs_name()
            role_name = self._query_role_name('DATANODE', host)
            role_names = {"items": [role_name]}
            self._role_command(self.cluster_name, hdfs_name, role_names, 'start')
            while self._role_status(self.cluster_name, hdfs_name, role_name) != 'STARTED':
                time.sleep(5)
    
        def stop_datanode(self, host):
            self._get_input_param()
            log('stopping datanode on %s' % host)
            hdfs_name = self._query_hdfs_name()
            role_name = self._query_role_name('DATANODE', host)
            role_names = {"items": [role_name]}
            self._role_command(self.cluster_name, hdfs_name, role_names, 'stop')
            while self._role_status(self.cluster_name, hdfs_name, role_name) != 'STOPPED':
                time.sleep(5)
    
        def query_datanode_dir(self, host):
            role_name = self._query_role_name("DATANODE", host)
            service_name = self._query_hdfs_name()
            return self._query_role_conf(self.cluster_name, service_name, role_name, "dfs_data_dir_list")
    
        def remove_datanode_dir(self, host, dir):
            role_name = self._query_role_name("DATANODE", host)
            service_name = self._query_hdfs_name()
            data_dir_list = self.query_datanode_dir(host)
            log('remove data dir: %s from dirs: %s' % (dir, data_dir_list))
            dirs = [os.path.abspath(x) for x in data_dir_list.split(",")]
            try:
                dirs.remove(os.path.abspath(dir))
                self._update_role_conf(self.cluster_name, service_name, role_name, "dfs_data_dir_list",
                                       ','.join(dirs))
            except ValueError:
                log("%s was not in %s" % (dir, data_dir_list))
    
        def discover_datanodes(self):
            """
            :return: list(DataNode)
            """
            self._get_input_param()
            result = []
            hdfs_name = self._query_hdfs_name()
            dn_role_groups = self._query_role_groups(self.cluster_name, hdfs_name, 'DATANODE')
            for group in dn_role_groups:
                hostnames = self._query_role_group_nodes(self.cluster_name, hdfs_name, group)
                data_dir = self._query_role_group_conf(self.cluster_name, hdfs_name, group, 'dfs_data_dir_list')
                data_nodes = [DataNode(hostname, group, data_dir) for hostname in hostnames]
                result += data_nodes
            return result
    
    
    class HDP(HadoopDist):
        def __init__(self):
            super(HDP, self).__init__()
            self.home = "/usr/hdp/current"
    
            # hive auxlib
            hive_aux_dir = None
            if os.path.exists(os.path.join(self.home, "hive-client")):
                hive_aux_dir = os.path.join(self.home, "hive-client/auxlib")
                if not os.path.exists(hive_aux_dir):
                    os.mkdir(hive_aux_dir)
    
            self.link_dirs = [os.path.join(self.home, app) for app in
                              ["hadoop-client/lib", "spark2-client/jars", "druid-overlord/extensions/druid-hdfs-storage",
                               hive_aux_dir] if app is not None]
            self.addr = None
    
        def _get_input_param(self):
            if self.addr is not None:
                return
            import socket
            local_address = socket.gethostname() + ":8080"
            address = raw_input("please input Ambari address(%s):" % local_address) or local_address
            hostname_port = address.split(":")
            while len(hostname_port) != 2:
                log("input wrong Ambari address, please input again")
                address = raw_input("please input Ambari address(%s):" % local_address) or local_address
                hostname_port = address.split(":")
            self.host = hostname_port[0].strip()
            self.port = hostname_port[1].strip()
            self.rest_api_url_base = "http://%s:%s/api/v1/" % (self.host, self.port)
            self.username = raw_input("please input Ambari admin user(admin):") or 'admin'
            self.password = raw_input("please input Ambari admin password(admin):") or 'admin'
            check_input(self.username, self.password, urljoin(self.rest_api_url_base, "clusters"))
            self.addr = "%s:%s@%s:%s" % (self.username, self.password, self.host, self.port)
            self.cluster_name = self._query_cluster_name()
    
        def _get_all_prop(self, config_type):
            url = urljoin(self.rest_api_url_base, "clusters/%s/?fields=Clusters/desired_configs" % self.cluster_name)
            response = self.do_get(url)
            desired_tags = json.loads(response)
            tag = desired_tags["Clusters"]["desired_configs"][config_type]["tag"]
            url = urljoin(self.rest_api_url_base,
                          "clusters/%s/configurations?type=%s&tag=%s" % (self.cluster_name, config_type, tag))
            response = self.do_get(url)
            items = json.loads(response)["items"]
            if len(items) > 0:
                return items[0]["properties"]
            else:
                return None
    
        def _get_prop(self, config_type, keys):
            conf_dic = self._get_all_prop(config_type)
            if conf_dic is not None:
                return [conf_dic[x] for x in keys]
            else:
                return None
    
        def _set_prop(self, config_type, key, value):
            cmd = "/var/lib/ambari-server/resources/scripts/configs.py -u %s -p %s -t %s -a set -l %s -n %s -c " \
                  "%s -k %s -v '%s' > /dev/null 2>&1" % (
                      self.username, self.password, self.port, self.host, self.cluster_name, config_type, key, value)
            code = os.system(cmd)
            if code > 0:
                log("failed to set %s:%s=%s" % (config_type, key, value))
    
        def query_configured_core_site(self):
            conf_dic = self._get_all_prop("core-site")
            return dict(filter(lambda x: is_jfs_conf(x[0]), conf_dic.items()))
    
        def query_configured_mapred_site(self):
            result = dict()
            result['mapreduce.application.classpath'] = self._get_prop("mapred-site",
                                                                       ["mapreduce.application.classpath"])[0]
            return result
    
        def update_core_site(self, conf_dict):
            if not os.path.exists("/var/lib/ambari-server/resources/scripts/configs.py"):
                log("config command can only run on ambari-server machine")
                sys.exit(1)
            for k, v in conf_dict.items():
                self._set_prop("core-site", k, v)
    
            if self._has_service("RANGER"):
                schemes = self._get_prop("ranger-hive-security", ["ranger.plugin.hive.urlauth.filesystem.schemes"])[0]
                if 'jfs' not in schemes:
                    schemes = schemes + ',jfs:'
                    self._set_prop("ranger-hive-security", "ranger.plugin.hive.urlauth.filesystem.schemes", schemes)
                    self.need_to_add_conf['ranger.plugin.hive.urlauth.filesystem.schemes'] = schemes
    
        def _has_service(self, service_name):
            url = urljoin(self.rest_api_url_base,
                          "clusters/%s/services" % self.cluster_name)
            response = self.do_get(url)
            items = json.loads(response)["items"]
            services = [x['ServiceInfo']['service_name'] for x in items]
            return service_name in services
    
        def update_mapred_site(self, conf_dict):
            cp = conf_dict['mapreduce.application.classpath']
            if cp and "juicefs-hadoop" not in cp:
                cp += ":%s" % self.jar_link_path
                self._set_prop("mapred-site", "mapreduce.application.classpath", cp)
                self.need_to_add_conf['mapreduce.application.classpath'] = cp
    
        def update_hive(self):
            hive_aux_jars = self._get_prop("hive-interactive-env", ["hive_aux_jars"])[0]
            if hive_aux_jars and "juicefs-hadoop.jar" not in hive_aux_jars:
                hive_aux_jars += ",file://%s" % os.path.join(self.download_dir, "juicefs-hadoop.jar")
            else:
                hive_aux_jars = "file://%s" % os.path.join(self.download_dir, "juicefs-hadoop.jar")
            self._set_prop("hive-interactive-env", "hive_aux_jars", hive_aux_jars)
    
        def deploy_config(self):
            self._get_input_param()
            hadoop_clients = ["HDFS_CLIENT"]
            for host in self.discover_nodes():
                url = urljoin(self.rest_api_url_base,
                              "clusters/%s/hosts/%s/host_components/" % (self.cluster_name, self.host))
                response = self.do_get(url)
                components = json.loads(response)["items"]
                clients = list(filter(lambda x: x["HostRoles"]["component_name"] in hadoop_clients, components))
    
                if len(clients) == 0:
                    continue
    
                def build_item(client):
                    service_name = client["HostRoles"]["component_name"].split("_")[0]
                    component_name = client["HostRoles"]["component_name"]
                    return {
                        "service_name": "%s" % service_name,
                        "component_name": "%s" % component_name,
                        "hosts": "%s" % host
                    }
    
                resource_filters = list(map(lambda x: build_item(x), clients))
    
                request_info = {
                    "command": "RESTART",
                    "context": "Restart all clients on %s" % host,
                    "operation_level": {
                        "level": "HOST",
                        "cluster_name": "%s" % self.cluster_name
                    }
                }
    
                url = urljoin(self.rest_api_url_base, "clusters/%s/requests" % self.cluster_name)
                data = {"RequestInfo": request_info, "Requests/resource_filters": resource_filters}
                header = {"X-Requested-By": "ambari"}
    
                code = self.do_post(url, json.dumps(data), header)
                assert code == 202
                log("restart all clients on %s success" % host)
    
        def do_discover_nodes(self):
            self._get_input_param()
            url = urljoin(self.rest_api_url_base, "clusters/%s/hosts" % self.cluster_name)
            response = self.do_get(url)
            items = json.loads(response)["items"]
            nodes = [x["Hosts"]["host_name"] for x in items]
            return nodes
    
        def _query_cluster_name(self):
            url = urljoin(self.rest_api_url_base, "clusters")
            response = self.do_get(url)
            clusters = json.loads(response)["items"]
            names = [x["Clusters"]["cluster_name"] for x in clusters]
            if len(clusters) == 1:
                cluster_name = names[0]
            elif len(clusters) > 1:
                cluster_name = raw_input("please specify a cluster(%s):" % ", ".join(names)) or names[0]
            else:
                log("cluster has not been created yet")
                sys.exit(1)
            return cluster_name
    
    
    class AmazonEMR(HadoopDist):
        def __init__(self):
            super(AmazonEMR, self).__init__()
            self.link_dirs = [
                "/usr/lib/hadoop/lib/",
                "/usr/lib/spark/jars/",
                "/usr/lib/hive/auxlib/"
            ]
    
    
    class AliyunEMR(HadoopDist):
        def __init__(self):
            super(AliyunEMR, self).__init__()
            self.core_site_path = "/etc/ecm/hadoop-conf/core-site.xml"
            self.mapred_site_path = "/etc/ecm/hadoop-conf/mapred-site.xml"
            dir_patterns = [
                "/opt/apps/ecm/service/hadoop/*/package/hadoop*/share/hadoop/common/lib",
                "/opt/apps/ecm/service/spark/*/package/spark*/jars",
                "/opt/apps/ecm/service/presto/*/package/presto*/plugin/hive-hadoop2",
                "/opt/apps/ecm/service/hive/*/package/apache-hive*/lib",
                "/opt/apps/ecm/service/impala/*/package/impala*/lib"
            ]
            self.link_dirs = [glob.glob(p)[0] for p in dir_patterns if len(p) > 0]
    
        def do_discover_nodes(self):
            return [x.split()[0] for x in open("/etc/hosts", "r").readlines() if "emr-worker" in x]
    
    
    class TencentEMR(HadoopDist):
        def __init__(self):
            super(TencentEMR, self).__init__()
            self.core_site_path = "/usr/local/service/hadoop/etc/hadoop/core-site.xml"
            self.mapred_site_path = "/usr/local/service/hadoop/etc/hadoop/mapred-site.xml"
            JUICEFS_CONF['juicefs.superuser'] = 'hadoop'
            self.hdfs_superuser = 'hadoop'
            self.home = "/usr/local/service/"
            self.link_dirs = [os.path.join(self.home, app) for app in ["presto/plugin/hive-hadoop2",
                                                                       "hadoop/share/hadoop/common/lib",
                                                                       "spark/jars",
                                                                       "hive/auxlib"]]
            self.hdfs_superuser = 'hadoop'
    
    
    class UHadoop(HadoopDist):
        def __init__(self):
            super(UHadoop, self).__init__()
            self.hdfs_superuser = 'hadoop'
            self.core_site_path = "/home/hadoop/conf/core-site.xml"
            self.mapred_site_path = "/home/hadoop/conf/mapred-site.xml"
            self.hdfs_superuser = 'hadoop'
            JUICEFS_CONF['juicefs.superuser'] = 'hadoop'
            self.home = "/home/hadoop"
            self.link_dirs = [os.path.join(self.home, app) for app in ["presto/plugin/hive-hadoop2",
                                                                       "share/hadoop/common/lib",
                                                                       "spark/jars",
                                                                       "hive/auxlib"]]
    
    
    def find_hadoop_dist():
        sts, text = commands.getstatusoutput('%s version' % HADOOP_CMD)
        if "cdh" in text and not os.path.exists("/home/hadoop/share/hadoop"):
            return CDH()
        if "hdp" in text:
            return HDP()
        if os.path.exists("/usr/local/service/hadoop"):
            return TencentEMR()
        if os.path.exists("/usr/lib/hadoop/lib/"):
            return AmazonEMR()
        if os.path.exists("/home/hadoop/share/hadoop"):
            return UHadoop()
        if os.path.exists("/opt/apps/ecm/service/hadoop"):
            return AliyunEMR()
        else:
            return HadoopDist()
    
    
    def usage():
        print("""python setup-hadoop.py COMMAND
    
    COMMAND could be:
      install         install JuiceFS jar on localhost
      install_all     install JuiceFS jar on all managed hosts
      config          auto add JuiceFS conf to Ambari or ClouderaManager
      deploy_config   deploy config to all managed hosts
      test            verify if JuiceFS is installed correctly on localhost
      test_all        verify if JuiceFS is installed correctly on all managed hosts
    """)
    
    
    COMMANDS = ["install", "uninstall", "config", "test", "install_all", "uninstall_all",
                "deploy_config", "test_all", "restart", "make_cache_dir",
                "shrink_datanode", "benchmark", "hive_meta", "hive_data", "hive_jfs"]
    
    
    def main():
        if len(sys.argv) < 2:
            return usage()
        dist = find_hadoop_dist()
        if len(sys.argv) == 1:
            dist.install()
            # dist.config()
            # dist.install_all()
            # dist.test()
        elif len(sys.argv) >= 2:
            command = sys.argv[1]
            if command in ['-h', '-H', '--help']:
                return usage()
            if command in COMMANDS:
                return getattr(dist, command)()
            else:
                print("not supported command")
                return 1
    
    
    sys.exit(main() or 0)

# #

## 未来改进 ##

在上面的缩容过程里，需要将数据是从一块磁盘完整地拷贝到另外一块磁盘，需要它有足够的剩余空间，另外也可能导致 DataNode 内磁盘间数据不均衡。未来可以改进下这个缩容过程，在复制数据时根据某个规则将 blk 文件复制到多块盘，确保多块磁盘之间的数据均衡。

原文:

[https://juicefs.com/blog/cn/posts/hdfs-datanode-shrink/][https_juicefs.com_blog_cn_posts_hdfs-datanode-shrink]

[Link 1]: https://hadoop.apache.org/docs/current/hadoop-yarn/hadoop-yarn-site/GracefulDecommission.html
[CDH]: https://docs.cloudera.com/documentation/enterprise/5-9-x/topics/cdh_ig_decommision_datanodes.html
[HDP]: https://docs.cloudera.com/HDPDocuments/HDP2/HDP-2.6.5/bk_administration/content/ref-a179736c-eb7c-4dda-b3b4-6f3a778bd8c8.1.html
[setup-hadoop.py]: http://s.juicefs.com/static/setup-hadoop.py
[https_juicefs.com_blog_cn_posts_hdfs-datanode-shrink]: https://juicefs.com/blog/cn/posts/hdfs-datanode-shrink/