~bigdata-dev/charm-helpers/framework

# Copyright 2014-2015 Canonical Limited.
#
# This file is part of charm-helpers.
#
# charm-helpers is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License version 3 as
# published by the Free Software Foundation.
#
# charm-helpers is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with charm-helpers.  If not, see <http://www.gnu.org/licenses/>.

import re
from subprocess import check_output, check_call
import time

from path import Path
import yaml

import jujuresources

from charmhelpers.core import host
from charmhelpers.core import hookenv
from charmhelpers.core import charmdata
from charmhelpers import fetch
from charmhelpers.contrib.bigdata import utils


class HadoopBase(object):
    JAVA_VERSION = '7'

    def __init__(self):
        self.cpu_arch = check_output(['uname', '-p']).strip()
        self.load_dist_config()
        self.spec = {
            'vendor': self.vendor,
            'hadoop': self.hadoop_version,
            'java': self.JAVA_VERSION,
            'arch': self.cpu_arch,
        }
        self.client_spec = {
            'hadoop': self.hadoop_version,
        }

    def load_dist_config(self, filename='dist.yaml'):
        self._dist_config = yaml.load(Path(filename).text())

        # validate dist.yaml
        required_opts = ['vendor', 'hadoop_version', 'dirs']
        optional_opts = {'packages': [], 'groups': [], 'users': {}, 'ports': {}}
        missing_opts = set(required_opts) - set(self._dist_config.keys())
        if missing_opts:
            raise ValueError('{} is missing required option{}: {}'.format(
                filename,
                's' if len(missing_opts) > 1 else '',
                ', '.join(missing_opts)))
        required_dirs = ['hadoop']
        missing_dirs = set(required_dirs) - set(self._dist_config['dirs'].keys())
        if missing_dirs:
            raise ValueError('dirs option in {} is missing required entr{}: {}'.format(
                filename,
                'ies' if len(missing_dirs) > 1 else 'y',
                ', '.join(missing_dirs)))

        for opt in required_opts:
            setattr(self, opt, self._dist_config[opt])
        for opt in optional_opts:
            setattr(self, opt, self._dist_config.get(opt, optional_opts[opt]))
        self.managed_dirs = self.dirs
        for name, details in self.managed_dirs.items():
            details['path'] = self._expand_path(details['path'])
        self.managed_dirs['hadoop_conf'] = {'path': Path('/etc/hadoop/conf')}
        self.dirs = {name: details['path'] for name, details in self.managed_dirs.items()}

    def _expand_path(self, path):
        return Path(path.format(config=hookenv.config()))

    def verify_conditional_resources(self):
        conditional_resources = ['hadoop-%s' % self.cpu_arch]
        if self.cpu_arch == 'ppc64le':
            conditional_resources.append('ibm-java-installer')
        mirror_url = hookenv.config()['resources_mirror']
        return jujuresources.fetch(conditional_resources, mirror_url=mirror_url)

    def is_installed(self):
        return charmdata.kv.get('hadoop.base.installed')

    def install(self, force=False):
        if not force and self.is_installed():
            return
        self.configure_hosts_file()
        self.manage_users()
        self.manage_dirs()
        self.install_base_packages()
        self.setup_hadoop_config()
        self.configure_hadoop()
        self.config_for_hue()
        charmdata.kv.set('hadoop.base.installed', True)

    def configure_hosts_file(self):
        """
        Add the unit's private-address to /etc/hosts to ensure that Java
        can resolve the hostname of the server to its real IP address.
        """
        private_address = hookenv.unit_get('private-address')
        hostname = check_output(['hostname']).strip()
        hostfqdn = check_output(['hostname', '-f']).strip()
        etc_hosts = Path('/etc/hosts')
        hosts = etc_hosts.lines()
        line = '%s %s %s' % (private_address, hostfqdn, hostname)
        IP_pat = r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}'
        if not re.match(IP_pat, private_address):
            line = '# %s  # private-address did not return an IP' % line
        if hosts[0] != line:
            hosts.insert(0, line)
            etc_hosts.write_lines(hosts)

    def manage_users(self):
        for group in self.groups:
            host.add_group(group)
        for username, details in self.users.items():
            primary_group = None
            groups = details.get('groups', [])
            if groups:
                primary_group = groups[0]
            host.adduser(username, group=primary_group)
            for group in groups:
                host.add_user_to_group(username, group)

    def manage_dirs(self):
        for name, details in self.managed_dirs.items():
            host.mkdir(
                self._expand_path(details['path']),
                owner=details.get('owner', 'root'),
                group=details.get('group', 'root'),
                perms=details.get('perms', 0o755))

    def install_base_packages(self):
        with utils.disable_firewall():
            fetch.apt_update()
            fetch.apt_install(self.packages)
            self.install_jdk()
            self.install_hadoop()

    def find_jdk(self):
        # ppc64le uses the ibm jdk; other arches use openjdk
        if self.cpu_arch == 'ppc64le':
            output = check_output(['find', '/opt/', '-name', 'java-ppc64le-*'])
        else:
            output = check_output(['find', '/usr/', '-name', 'java-%s-openjdk-*' % self.JAVA_VERSION])
        if output.strip():
            return Path(output.split('\n')[0])
        else:
            return None

    def install_jdk(self):
        if self.find_jdk():
            return
        if self.cpu_arch == 'ppc64le':
            java_installer = Path(jujuresources.resource_path('ibm-java-installer'))
            java_installer.chmod(0o755)
            check_call([java_installer, '-i', 'silent'])
        else:
            fetch.apt_install(['openjdk-%s-jdk' % self.JAVA_VERSION])

    def install_hadoop(self):
        jujuresources.install('hadoop-%s' % self.cpu_arch, destination=self.dirs['hadoop'], skip_top_level=True)

    def setup_hadoop_config(self):
        # copy default config into alternate dir
        conf_dir = self.dirs['hadoop'] / 'etc/hadoop'
        self.dirs['hadoop_conf'].rmtree_p()
        conf_dir.copytree(self.dirs['hadoop_conf'])
        (self.dirs['hadoop_conf'] / 'slaves').remove_p()
        mapred_site = self.dirs['hadoop_conf'] / 'mapred-site.xml'
        if not mapred_site.exists():
            (self.dirs['hadoop_conf'] / 'mapred-site.xml.template').copy(mapred_site)

    def configure_hadoop(self):
        config = hookenv.config()
        jdk_path = self.find_jdk()
        if not jdk_path:
            raise ValueError('Unable to find JDK')

        jdk_bin = jdk_path / 'bin'
        hadoop_bin = self.dirs['hadoop'] / 'bin'
        hadoop_sbin = self.dirs['hadoop'] / 'sbin'
        with utils.environment_edit_in_place('/etc/environment') as env:
            env['JAVA_HOME'] = jdk_path
            if jdk_bin not in env['PATH']:
                env['PATH'] = ':'.join([env['PATH'], jdk_bin])
            if hadoop_bin not in env['PATH']:
                env['PATH'] = ':'.join([env['PATH'], hadoop_bin])
            if hadoop_sbin not in env['PATH']:
                env['PATH'] = ':'.join([env['PATH'], hadoop_sbin])
            env['HADOOP_LIBEXEC_DIR'] = self.dirs['hadoop'] / 'libexec'
            env['HADOOP_INSTALL'] = self.dirs['hadoop']
            env['HADOOP_HOME'] = self.dirs['hadoop']
            env['HADOOP_COMMON_HOME'] = self.dirs['hadoop']
            env['HADOOP_HDFS_HOME'] = self.dirs['hadoop']
            env['HADOOP_MAPRED_HOME'] = self.dirs['hadoop']
            env['HADOOP_YARN_HOME'] = self.dirs['hadoop']
            env['YARN_HOME'] = self.dirs['hadoop']
            env['HADOOP_CONF_DIR'] = self.dirs['hadoop_conf']
            env['YARN_CONF_DIR'] = self.dirs['hadoop_conf']
            env['YARN_LOG_DIR'] = self.dirs['yarn_log_dir']
            env['HDFS_LOG_DIR'] = self.dirs['hdfs_log_dir']
            env['HADOOP_LOG_DIR'] = self.dirs['hdfs_log_dir']  # for hadoop 2.2.0 only
            env['MAPRED_LOG_DIR'] = '/var/log/hadoop/mapred'  # should be moved to config, but could
            env['MAPRED_PID_DIR'] = '/var/run/hadoop/mapred'  # be destructive for mapreduce operation

        hadoop_env = self.dirs['hadoop_conf'] / 'hadoop-env.sh'
        utils.re_edit_in_place(hadoop_env, {
            r'export JAVA_HOME *=.*': 'export JAVA_HOME=%s' % jdk_path,
        })

        core_site = self.dirs['hadoop_conf'] / 'core-site.xml'
        with utils.xmlpropmap_edit_in_place(core_site) as props:
            props["io.file.buffer.size"] = config["io_file_buffer_size"]

    def config_for_hue(self):
        # Set hue-specific properties in our hdfs|core-site.xml files
        hdfs_site = self.dirs['hadoop_conf'] / 'hdfs-site.xml'
        with utils.xmlpropmap_edit_in_place(hdfs_site) as props:
            props['dfs.webhdfs.enabled'] = "true"
        core_site = self.dirs['hadoop_conf'] / 'core-site.xml'
        with utils.xmlpropmap_edit_in_place(core_site) as props:
            props['hadoop.proxyuser.hue.hosts'] = "*"
            props['hadoop.proxyuser.hue.groups'] = "*"

    def run(self, user, command, *args, **kwargs):
        """
        Run a Hadoop command as the `hdfs` user.

        :param str command: Command to run, prefixed with `bin/` or `sbin/`
        :param list args: Additional args to pass to the command
        """
        return utils.run_as(user, self.dirs['hadoop'] / command, *args, **kwargs)


class HDFS(object):
    def __init__(self, hadoop_base):
        self.hadoop_base = hadoop_base
        self.master_ports = hadoop_base.ports['hdfs']['master']
        self.secondarynamenode_ports = hadoop_base.ports['hdfs']['secondarynamenode']
        self.slave_ports = hadoop_base.ports['hdfs']['slave']

    def stop_namenode(self):
        self._hadoop_daemon('stop', 'namenode')

    def start_namenode(self):
        if not utils.jps('NameNode'):
            self._hadoop_daemon('start', 'namenode')
        # Some hadoop processes take a bit of time to start
        # we need to let them get to a point where they are
        # ready to accept connections - increase the value for hadoop 2.4.1
        time.sleep(30)

    def stop_secondarynamenode(self):
        self._hadoop_daemon('stop', 'secondarynamenode')

    def start_secondarynamenode(self):
        if not utils.jps('SecondaryNameNode'):
            self._hadoop_daemon('start', 'secondarynamenode')
        # Some hadoop processes take a bit of time to start
        # we need to let them get to a point where they are
        # ready to accept connections - increase the value for hadoop 2.4.1
        time.sleep(30)

    def stop_datanode(self):
        self._hadoop_daemon('stop', 'datanode')

    def start_datanode(self):
        self._hadoop_daemon('start', 'datanode')

    def _remote_host(self):
        rels = charmdata.kv.get('relations.ready')
        namenode = rels.get('namenode', rels.get('datanode'))
        return namenode.values()[0]['private-address']

    def _local_host(self):
        return hookenv.unit_get('private-address')

    def configure_namenode(self):
        if not charmdata.kv.get('hdfs.namenode.configured'):
            self.configure_hdfs_base(self._local_host())
            self.format_namenode()
            self.create_hdfs_dirs()
            charmdata.kv.set('hdfs.namenode.configured', True)

    def configure_secondarynamenode(self):
        """
        Configure the Secondary Namenode when the apache-hadoop-hdfs-checkpoint
        charm is deployed and related to apache-hadoop-hdfs-master.

        The only purpose of the secondary namenode is to perform periodic
        checkpoints. The secondary name-node periodically downloads current
        namenode image and edits log files, joins them into new image and
        uploads the new image back to the (primary and the only) namenode.
        """
        if not charmdata.kv.get('hdfs.secondarynamenode.configured'):
            local = self._local_host()
            remote = self._remote_host()
            # Set generic hdfs config
            self.configure_hdfs_base(remote)
            # Set secondarynamenode-specific config
            hdfs_site = self.hadoop_base.dirs['hadoop_conf'] / 'hdfs-site.xml'
            with utils.xmlpropmap_edit_in_place(hdfs_site) as props:
                props['dfs.namenode.http-address'] = "{host}:50070".format(host=remote)
                props['dfs.namenode.secondary.http-address'] = "{host}:50090".format(host=local)
            self.start_secondarynamenode()
            charmdata.kv.set('hdfs.secondarynamenode.configured', True)

    def configure_datanode(self):
        if not charmdata.kv.get('hdfs.datanode.configured'):
            self.configure_hdfs_base(self._remote_host())
            self.start_datanode()
            charmdata.kv.set('hdfs.datanode.configured', True)

    def configure_client(self):
        if not charmdata.kv.get('hdfs.client.configured'):
            self.configure_hdfs_base(self._remote_host())
            charmdata.kv.set('hdfs.client.configured', True)

    def configure_hdfs_base(self, host):
        config = hookenv.config()
        core_site = self.hadoop_base.dirs['hadoop_conf'] / 'core-site.xml'
        with utils.xmlpropmap_edit_in_place(core_site) as props:
            props['fs.defaultFS'] = "hdfs://{host}:8020".format(host=host)
        hdfs_site = self.hadoop_base.dirs['hadoop_conf'] / 'hdfs-site.xml'
        with utils.xmlpropmap_edit_in_place(hdfs_site) as props:
            props['dfs.namenode.name.dir'] = Path(config['hadoop_dir_base']) / 'cache/hadoop/dfs/name'
            props['dfs.namenode.handler.count'] = config['dfs_namenode_handler_count']
            props['dfs.datanode.max.transfer.threads'] = config['dfs_datanode_max_xcievers']
            props['dfs.datanode.data.dir'] = Path(config['hadoop_dir_base']) / 'cache/hadoop/dfs/name'
            props['dfs.permissions'] = 'false'  # TODO - secure this hadoop installation!
            props['dfs.replication'] = config['dfs_replication']
            props['dfs.heartbeat.interval'] = config['dfs_heartbeat_interval']
            props['dfs.namenode.heartbeat.recheck-interval'] = config['dfs_namenode_heartbeat_recheck_interval']

    def format_namenode(self):
        self.stop_namenode()
        self._hdfs('namenode', '-format', '-force')
        self.start_namenode()

    def create_hdfs_dirs(self):
        self._hdfs('dfs', '-mkdir', '-p', '/tmp/hadoop/mapred/staging')
        self._hdfs('dfs', '-chmod', '-R', '1777', '/tmp/hadoop/mapred/staging')
        self._hdfs('dfs', '-mkdir', '-p', '/tmp/hadoop-yarn/staging')
        self._hdfs('dfs', '-chmod', '-R', '1777', '/tmp/hadoop-yarn')
        self._hdfs('dfs', '-mkdir', '-p', '/user/ubuntu')
        self._hdfs('dfs', '-chown', '-R', 'ubuntu', '/user/ubuntu')
        self._hdfs('dfs', '-mkdir', '-p', '/user/hue')
        self._hdfs('dfs', '-chown', '-R', 'hue', '/user/hue')
        # for JobHistory
        self._hdfs('dfs', '-mkdir', '-p', '/mr-history/tmp')
        self._hdfs('dfs', '-chmod', '-R', '1777', '/mr-history/tmp')
        self._hdfs('dfs', '-mkdir', '-p', '/mr-history/done')
        self._hdfs('dfs', '-chmod', '-R', '1777', '/mr-history/done')
        self._hdfs('dfs', '-chown', '-R', 'mapred:hdfs', '/mr-history')
        self._hdfs('dfs', '-mkdir', '-p', '/app-logs')
        self._hdfs('dfs', '-chmod', '-R', '1777', '/app-logs')
        self._hdfs('dfs', '-chown', 'yarn', '/app-logs')

    def register_slaves(self):
        slaves = charmdata.kv.get('relations.ready', {}).get('datanode', {})
        slaves_file = self.hadoop_base.dirs['hadoop_conf'] / 'slaves'
        slaves_file.write_lines(
            [
                '# DO NOT EDIT',
                '# This file is automatically managed by Juju',
            ] + [
                slave['private-address'] for slave in slaves.values()
            ]
        )
        slaves_file.chown('ubuntu', 'hadoop')

    def _hadoop_daemon(self, command, service):
        self.hadoop_base.run('hdfs', 'sbin/hadoop-daemon.sh',
                             '--config', self.hadoop_base.dirs['hadoop_conf'],
                             command, service)

    def _hdfs(self, command, *args):
        self.hadoop_base.run('hdfs', 'bin/hdfs', command, *args)


class YARN(object):
    def __init__(self, hadoop_base):
        self.hadoop_base = hadoop_base
        self.master_ports = hadoop_base.ports['yarn']['master']
        self.slave_ports = hadoop_base.ports['yarn']['slave']

    def stop_resourcemanager(self):
        self._yarn_daemon('stop', 'resourcemanager')

    def start_resourcemanager(self):
        if not utils.jps('ResourceManager'):
            self._yarn_daemon('start', 'resourcemanager')

    def stop_jobhistory(self):
        self._jobhistory_daemon('stop', 'historyserver')

    def start_jobhistory(self):
        if utils.jps('JobHistoryServer'):
            self._jobhistory_daemon('stop', 'historyserver')
        self._jobhistory_daemon('start', 'historyserver')

    def stop_nodemanager(self):
        self._yarn_daemon('stop', 'nodemanager')

    def start_nodemanager(self):
        if not utils.jps('NodeManager'):
            self._yarn_daemon('start', 'nodemanager')

    def _remote_host(self):
        rels = charmdata.kv.get('relations.ready')
        resourcemanager = rels.get('resourcemanager', rels.get('nodemanager'))
        return resourcemanager.values()[0]['private-address']

    def _local_host(self):
        return hookenv.unit_get('private-address')

    def configure_resourcemanager(self):
        if not charmdata.kv.get('yarn.resourcemanager.configured'):
            host = self._local_host()
            self.configure_yarn_base(host)
            yarn_site = self.hadoop_base.dirs['hadoop_conf'] / 'yarn-site.xml'
            with utils.xmlpropmap_edit_in_place(yarn_site) as props:
                props["yarn.nodemanager.address"] = "{host}:8050".format(host=host)
                props["yarn.nodemanager.localizer.address"] = "{host}:8045".format(host=host)
            charmdata.kv.set('yarn.resourcemanager.configured', True)
            self.start_resourcemanager()

    def configure_jobhistory(self):
        if not charmdata.kv.get('yarn.jobhistory.configured'):
            host = self._local_host()
            self.configure_yarn_base(host)
            charmdata.kv.set('yarn.jobhistory.configured', True)
            self.start_jobhistory()

    def configure_nodemanager(self):
        if not charmdata.kv.get('yarn.nodemanager.configured'):
            self.configure_yarn_base(self._remote_host())
            charmdata.kv.set('yarn.nodemanager.configured', True)
            self.start_nodemanager()

    def configure_client(self):
        if not charmdata.kv.get('yarn.client.configured'):
            self.configure_yarn_base(self._remote_host())
            self.install_demo()
            charmdata.kv.set('yarn.client.configured', True)

    def configure_yarn_base(self, host):
        config = hookenv.config()
        yarn_site = self.hadoop_base.dirs['hadoop_conf'] / 'yarn-site.xml'
        with utils.xmlpropmap_edit_in_place(yarn_site) as props:
            props["yarn.resourcemanager.resource-tracker.address"] = "{host}:8031".format(host=host)
            props["yarn.resourcemanager.scheduler.address"] = "{host}:8030".format(host=host)
            props["yarn.resourcemanager.address"] = "{host}:8032".format(host=host)
            props["yarn.log.server.url"] = "{host}:19888/jobhistory/logs/".format(host=host)
            props["yarn.resourcemanager.webapp.address"] = "{host}:8088".format(host=host)
            props["yarn.nodemanager.aux-services"] = config["yarn_nodemanager_aux-services"]
            props["yarn.nodemanager.aux-services.mapreduce.shuffle.class"] = \
                config["yarn_nodemanager_aux-services_mapreduce_shuffle_class"]
        mapred_site = self.hadoop_base.dirs['hadoop_conf'] / 'mapred-site.xml'
        with utils.xmlpropmap_edit_in_place(mapred_site) as props:
            props["mapreduce.jobhistory.address"] = "{host}:10020".format(host=host)
            props["mapreduce.jobhistory.webapp.address"] = "{host}:19888".format(host=host)
            props["mapreduce.framework.name"] = config["mapreduce_framework_name"]
            props["mapreduce.reduce.shuffle.parallelcopies"] = config["mapreduce_reduce_shuffle_parallelcopies"]
            props["mapred.child.java.opts"] = config["mapred_child_java_opts"]
            props["mapreduce.task.io.sort.factor"] = config["mapreduce_task_io_sort_factor"]
            props["mapreduce.task.io.sort.mb"] = config["mapreduce_task_io_sort_mb"]
            props["mapred.job.tracker.handler.count"] = config["mapred_job_tracker_handler_count"]
            props["tasktracker.http.threads"] = config["tasktracker_http_threads"]

    def install_demo(self):
        # Copy our demo (TeraSort) to the target location and set mode/owner
        demo_source = 'scripts/terasort.sh'
        demo_target = '/home/ubuntu/terasort.sh'

        Path(demo_source).copy(demo_target)
        Path(demo_target).chmod(0o755)
        Path(demo_target).chown('ubuntu', 'hadoop')

    def _yarn_daemon(self, command, service):
        self.hadoop_base.run('yarn', 'sbin/yarn-daemon.sh',
                             '--config', self.hadoop_base.dirs['hadoop_conf'],
                             command, service)

    def _jobhistory_daemon(self, command, service):
        # TODO refactor job history to separate class
        self.hadoop_base.run('mapred', 'sbin/mr-jobhistory-daemon.sh',
                             '--config', self.hadoop_base.dirs['hadoop_conf'],
                             command, service)