Merge lp:~bigdata-dev/charms/trusty/hdp-hadoop/trunk into lp:charms/trusty/hdp-hadoop

[Cory Johns]

Updates to support HDP 2.2

[amir sanjar (asanjar)]

review: Approve

[lp:~bigdata-dev/charms/trusty/hdp-hadoop/trunk]

41. By Cory Johns on 2015-02-13

Fix jps bug (and un-vendor charmhelpers)

Unmerged revisions

41. By Cory Johns on 2015-02-13

Fix jps bug (and un-vendor charmhelpers)

40. By amir sanjar on 2014-12-26

fixes for enabling hdp 2.2 support

39. By Cory Johns on 2014-12-17

Fixed mapreduce TODO config items and refactored setHadoopConfigXML

38. By Cory Johns on 2014-12-17

Cleaned up mapreduce test case

37. By Cory Johns on 2014-12-17

Refactor test case and fix "exit on PASS" bug in test

36. By Cory Johns on 2014-12-17

Default memory settings

35. By Cory Johns on 2014-12-15

Fixed hostname yarn property and cleaned up test case

34. By Cory Johns on 2014-12-15

Fixed mapreduce paths

33. By Cory Johns on 2014-12-15

Fixed yarn paths

32. By Cory Johns on 2014-12-15

Fixed TODO-JDK-PATH placeholder from upstream config

=== modified file 'README.md'
--- README.md	2014-09-10 18:29:03 +0000
+++ README.md	2015-02-13 00:38:49 +0000
@@ -1,5 +1,5 @@
 ## Overview
-**What is Hortonworks Apache Hadoop (HDP 2.1.3) ?**
+**What is Hortonworks Apache Hadoop (HDP 2.2.0) ?**
 The Apache Hadoop software library is a framework that allows for the
 distributed processing of large data sets across clusters of computers
 using a simple programming model.
@@ -10,7 +10,7 @@
 at the application layer, so delivering a highly-availabile service on top of a
 cluster of computers, each of which may be prone to failures.
 
-Apache Hadoop 2.4.1 consists of significant improvements over the previous
+Apache Hadoop 2.6.x consists of significant improvements over the previous
 stable release (hadoop-1.x).
 
 Here is a short overview of the improvments to both HDFS and MapReduce.
@@ -51,7 +51,7 @@
 
 This supports deployments of Hadoop in a number of configurations.
 
-### HDP 2.1.3 Usage #1: Combined HDFS and MapReduce
+### HDP 2.2.0 Usage #1: Combined HDFS and MapReduce
 
 In this configuration, the YARN ResourceManager is deployed on the same
 service units as HDFS namenode and the HDFS datanodes also run YARN NodeManager::
 
=== added file 'files/jujuresources-0.2.3.tar.gz'
 Binary files files/jujuresources-0.2.3.tar.gz	1970-01-01 00:00:00 +0000 and files/jujuresources-0.2.3.tar.gz	2015-02-13 00:38:49 +0000 differ
=== added file 'files/scripts/terasort.sh'
--- files/scripts/terasort.sh	1970-01-01 00:00:00 +0000
+++ files/scripts/terasort.sh	2015-02-13 00:38:49 +0000
@@ -0,0 +1,15 @@
+#!/bin/bash
+
+
+SIZE=10000
+NUM_MAPS=100
+NUM_REDUCES=100
+IN_DIR=in_dir
+OUT_DIR=out_dir
+hadoop fs -rm -r -skipTrash ${IN_DIR} || true
+hadoop jar /usr/hdp/current/hadoop-mapreduce-client/hadoop-mapreduce-examples.jar teragen ${SIZE} ${IN_DIR}
+
+sleep 20
+
+hadoop jar /usr/hdp/current/hadoop-mapreduce-client/hadoop-mapreduce-examples.jar terasort ${IN_DIR} ${OUT_DIR}
+
 
=== modified file 'hooks/bdutils.py'
--- hooks/bdutils.py	2014-09-17 14:49:16 +0000
+++ hooks/bdutils.py	2015-02-13 00:38:49 +0000
@@ -1,11 +1,15 @@
 #!/usr/bin/python
 import os
 import pwd
+import sys
 import grp
 import subprocess
+import hashlib
+import shlex
 
 from shutil import rmtree
 from charmhelpers.core.hookenv import log
+from charmhelpers.contrib.bigdata.utils import jps
 
 def createPropertyElement(name, value):
     import xml.etree.ElementTree as ET
@@ -22,26 +26,21 @@
     import xml.dom.minidom as minidom
     print("==> setHadoopConfigXML ","INFO")
     import xml.etree.ElementTree as ET
-    found = False
     with open(xmlfileNamePath,'rb+') as f:    
         root = ET.parse(f).getroot()
-        proList = root.findall("property")
-        for p in proList:
-            if found:
+        for p in root.findall("property"):
+            if p.find('name').text == name:
+                if value is None:
+                    root.remove(p)
+                else:
+                    p.find('value').text = value
                 break
-            cList = p.getchildren()
-            for c in cList:
-                if c.text == name:
-                    p.find("value").text = value
-                    found = True
-                    break
+        else:
+            if value is not None:
+                root.append(createPropertyElement(name, value))
 
         f.seek(0)
-        if not found:            
-            root.append(createPropertyElement(name, value))
-            f.write((minidom.parseString(ET.tostring(root, encoding='UTF-8'))).toprettyxml(indent="\t"))
-        else:
-            f.write(ET.tostring(root, encoding='UTF-8'))
+        f.write(ET.tostring(root, encoding='UTF-8'))
         f.truncate()
 
 def setDirPermission(path, owner, group, access):
@@ -67,15 +66,21 @@
     for r,d,f in os.walk(path):
         os.chmod( r , mode)
             
-def wgetPkg(pkgName, crypType):
+def wgetPkg(url, dest):
     log("==> wgetPkg ")
-    crypFileName= pkgName+'.'+crypType    
-    cmd = 'wget '+pkgName
-    subprocess.call(cmd.split())
-    if crypType:
-        cmd = ['wget', crypFileName]
-        subprocess.call(cmd)
-    #TODO -- cryption validation
+    output = subprocess.check_output(['wget', '-S', url, '-O', dest],
+                                     stderr=subprocess.STDOUT)
+    etag = [l for l in output.split('\n') if l.startswith('  ETag')]
+    if etag:
+        etag = etag[0].split(':')[1].strip(' "')
+        if '-' in etag:
+            log.warn('Multi-part ETag not supported; cannot verify checksum')
+            return  # TODO: Support multi-part ETag algorithm
+        with open(dest) as fp:
+            md5 = hashlib.md5(fp.read()).hexdigest()
+        if md5 != etag:
+            log.error('Checksum mismatch: %s != %s' % (md5, etag))
+            sys.exit(1)
      
 def append_bashrc(line):
     log("==> append_bashrc","INFO")
@@ -124,10 +129,21 @@
                 os.environ[ll[0]] = ll[1].strip().strip(';').strip("\"").strip()
                 
 def is_jvm_service_active(processname):
-    cmd=["jps"]
-    p = subprocess.Popen(cmd, stdout=subprocess.PIPE)
-    out, err = p.communicate()
-    if err == None and str(out).find(processname) != -1:
+    return jps(processname)
+
+
+def HDFS_command(command):
+    cmd = shlex.split("su hdfs -c 'hdfs {c}'".format(c=command))
+    return subprocess.check_output(cmd)
+
+def fconfigured(filename):
+    fpath = os.path.join(os.path.sep, 'home', 'ubuntu', filename)
+    if os.path.isfile(fpath):
         return True
     else:
-        return False
 \ No newline at end of file
+        touch(fpath)
+        return False      
+    
+def touch(fname, times=None):
+    with open(fname, 'a'):
+        os.utime(fname, times)
 
=== added file 'hooks/bootstrap.py'
--- hooks/bootstrap.py	1970-01-01 00:00:00 +0000
+++ hooks/bootstrap.py	2015-02-13 00:38:49 +0000
@@ -0,0 +1,9 @@
+import subprocess
+
+
+def install_charmhelpers():
+    subprocess.check_call(['apt-get', 'install', '-yq', 'python-pip'])
+    subprocess.check_call(['pip', 'install', 'files/jujuresources-0.2.3.tar.gz'])
+    import jujuresources
+    jujuresources.fetch()
+    jujuresources.install(['pathlib', 'pyaml', 'six', 'charmhelpers'])
 
=== removed directory 'hooks/charmhelpers'
=== removed file 'hooks/charmhelpers/__init__.py'
=== removed directory 'hooks/charmhelpers/core'
=== removed file 'hooks/charmhelpers/core/__init__.py'
=== removed file 'hooks/charmhelpers/core/fstab.py'
--- hooks/charmhelpers/core/fstab.py	2014-07-09 12:37:36 +0000
+++ hooks/charmhelpers/core/fstab.py	1970-01-01 00:00:00 +0000
@@ -1,114 +0,0 @@
-#!/usr/bin/env python
-# -*- coding: utf-8 -*-
-
-__author__ = 'Jorge Niedbalski R. <[e-mail]>'
-
-import os
-
-
-class Fstab(file):
-    """This class extends file in order to implement a file reader/writer
-    for file `/etc/fstab`
-    """
-
-    class Entry(object):
-        """Entry class represents a non-comment line on the `/etc/fstab` file
-        """
-        def __init__(self, device, mountpoint, filesystem,
-                     options, d=0, p=0):
-            self.device = device
-            self.mountpoint = mountpoint
-            self.filesystem = filesystem
-
-            if not options:
-                options = "defaults"
-
-            self.options = options
-            self.d = d
-            self.p = p
-
-        def __eq__(self, o):
-            return str(self) == str(o)
-
-        def __str__(self):
-            return "{} {} {} {} {} {}".format(self.device,
-                                              self.mountpoint,
-                                              self.filesystem,
-                                              self.options,
-                                              self.d,
-                                              self.p)
-
-    DEFAULT_PATH = os.path.join(os.path.sep, 'etc', 'fstab')
-
-    def __init__(self, path=None):
-        if path:
-            self._path = path
-        else:
-            self._path = self.DEFAULT_PATH
-        file.__init__(self, self._path, 'r+')
-
-    def _hydrate_entry(self, line):
-        return Fstab.Entry(*filter(
-            lambda x: x not in ('', None),
-            line.strip("\n").split(" ")))
-
-    @property
-    def entries(self):
-        self.seek(0)
-        for line in self.readlines():
-            try:
-                if not line.startswith("#"):
-                    yield self._hydrate_entry(line)
-            except ValueError:
-                pass
-
-    def get_entry_by_attr(self, attr, value):
-        for entry in self.entries:
-            e_attr = getattr(entry, attr)
-            if e_attr == value:
-                return entry
-        return None
-
-    def add_entry(self, entry):
-        if self.get_entry_by_attr('device', entry.device):
-            return False
-
-        self.write(str(entry) + '\n')
-        self.truncate()
-        return entry
-
-    def remove_entry(self, entry):
-        self.seek(0)
-
-        lines = self.readlines()
-
-        found = False
-        for index, line in enumerate(lines):
-            if not line.startswith("#"):
-                if self._hydrate_entry(line) == entry:
-                    found = True
-                    break
-
-        if not found:
-            return False
-
-        lines.remove(line)
-
-        self.seek(0)
-        self.write(''.join(lines))
-        self.truncate()
-        return True
-
-    @classmethod
-    def remove_by_mountpoint(cls, mountpoint, path=None):
-        fstab = cls(path=path)
-        entry = fstab.get_entry_by_attr('mountpoint', mountpoint)
-        if entry:
-            return fstab.remove_entry(entry)
-        return False
-
-    @classmethod
-    def add(cls, device, mountpoint, filesystem, options=None, path=None):
-        return cls(path=path).add_entry(Fstab.Entry(device,
-                                                    mountpoint, filesystem,
-                                                    options=options))
 
=== removed file 'hooks/charmhelpers/core/hookenv.py'
--- hooks/charmhelpers/core/hookenv.py	2014-07-09 12:37:36 +0000
+++ hooks/charmhelpers/core/hookenv.py	1970-01-01 00:00:00 +0000
@@ -1,498 +0,0 @@
-"Interactions with the Juju environment"
-# Copyright 2013 Canonical Ltd.
-#
-# Authors:
-#  Charm Helpers Developers <[e-mail]>
-
-import os
-import json
-import yaml
-import subprocess
-import sys
-import UserDict
-from subprocess import CalledProcessError
-
-CRITICAL = "CRITICAL"
-ERROR = "ERROR"
-WARNING = "WARNING"
-INFO = "INFO"
-DEBUG = "DEBUG"
-MARKER = object()
-
-cache = {}
-
-
-def cached(func):
-    """Cache return values for multiple executions of func + args
-
-    For example:
-
-        @cached
-        def unit_get(attribute):
-            pass
-
-        unit_get('test')
-
-    will cache the result of unit_get + 'test' for future calls.
-    """
-    def wrapper(*args, **kwargs):
-        global cache
-        key = str((func, args, kwargs))
-        try:
-            return cache[key]
-        except KeyError:
-            res = func(*args, **kwargs)
-            cache[key] = res
-            return res
-    return wrapper
-
-
-def flush(key):
-    """Flushes any entries from function cache where the
-    key is found in the function+args """
-    flush_list = []
-    for item in cache:
-        if key in item:
-            flush_list.append(item)
-    for item in flush_list:
-        del cache[item]
-
-
-def log(message, level=None):
-    """Write a message to the juju log"""
-    command = ['juju-log']
-    if level:
-        command += ['-l', level]
-    command += [message]
-    subprocess.call(command)
-
-
-class Serializable(UserDict.IterableUserDict):
-    """Wrapper, an object that can be serialized to yaml or json"""
-
-    def __init__(self, obj):
-        # wrap the object
-        UserDict.IterableUserDict.__init__(self)
-        self.data = obj
-
-    def __getattr__(self, attr):
-        # See if this object has attribute.
-        if attr in ("json", "yaml", "data"):
-            return self.__dict__[attr]
-        # Check for attribute in wrapped object.
-        got = getattr(self.data, attr, MARKER)
-        if got is not MARKER:
-            return got
-        # Proxy to the wrapped object via dict interface.
-        try:
-            return self.data[attr]
-        except KeyError:
-            raise AttributeError(attr)
-
-    def __getstate__(self):
-        # Pickle as a standard dictionary.
-        return self.data
-
-    def __setstate__(self, state):
-        # Unpickle into our wrapper.
-        self.data = state
-
-    def json(self):
-        """Serialize the object to json"""
-        return json.dumps(self.data)
-
-    def yaml(self):
-        """Serialize the object to yaml"""
-        return yaml.dump(self.data)
-
-
-def execution_environment():
-    """A convenient bundling of the current execution context"""
-    context = {}
-    context['conf'] = config()
-    if relation_id():
-        context['reltype'] = relation_type()
-        context['relid'] = relation_id()
-        context['rel'] = relation_get()
-    context['unit'] = local_unit()
-    context['rels'] = relations()
-    context['env'] = os.environ
-    return context
-
-
-def in_relation_hook():
-    """Determine whether we're running in a relation hook"""
-    return 'JUJU_RELATION' in os.environ
-
-
-def relation_type():
-    """The scope for the current relation hook"""
-    return os.environ.get('JUJU_RELATION', None)
-
-
-def relation_id():
-    """The relation ID for the current relation hook"""
-    return os.environ.get('JUJU_RELATION_ID', None)
-
-
-def local_unit():
-    """Local unit ID"""
-    return os.environ['JUJU_UNIT_NAME']
-
-
-def remote_unit():
-    """The remote unit for the current relation hook"""
-    return os.environ['JUJU_REMOTE_UNIT']
-
-
-def service_name():
-    """The name service group this unit belongs to"""
-    return local_unit().split('/')[0]
-
-
-def hook_name():
-    """The name of the currently executing hook"""
-    return os.path.basename(sys.argv[0])
-
-
-class Config(dict):
-    """A Juju charm config dictionary that can write itself to
-    disk (as json) and track which values have changed since
-    the previous hook invocation.
-
-    Do not instantiate this object directly - instead call
-    ``hookenv.config()``
-
-    Example usage::
-
-        >>> # inside a hook
-        >>> from charmhelpers.core import hookenv
-        >>> config = hookenv.config()
-        >>> config['foo']
-        'bar'
-        >>> config['mykey'] = 'myval'
-        >>> config.save()
-
-
-        >>> # user runs `juju set mycharm foo=baz`
-        >>> # now we're inside subsequent config-changed hook
-        >>> config = hookenv.config()
-        >>> config['foo']
-        'baz'
-        >>> # test to see if this val has changed since last hook
-        >>> config.changed('foo')
-        True
-        >>> # what was the previous value?
-        >>> config.previous('foo')
-        'bar'
-        >>> # keys/values that we add are preserved across hooks
-        >>> config['mykey']
-        'myval'
-        >>> # don't forget to save at the end of hook!
-        >>> config.save()
-
-    """
-    CONFIG_FILE_NAME = '.juju-persistent-config'
-
-    def __init__(self, *args, **kw):
-        super(Config, self).__init__(*args, **kw)
-        self._prev_dict = None
-        self.path = os.path.join(charm_dir(), Config.CONFIG_FILE_NAME)
-        if os.path.exists(self.path):
-            self.load_previous()
-
-    def load_previous(self, path=None):
-        """Load previous copy of config from disk so that current values
-        can be compared to previous values.
-
-        :param path:
-
-            File path from which to load the previous config. If `None`,
-            config is loaded from the default location. If `path` is
-            specified, subsequent `save()` calls will write to the same
-            path.
-
-        """
-        self.path = path or self.path
-        with open(self.path) as f:
-            self._prev_dict = json.load(f)
-
-    def changed(self, key):
-        """Return true if the value for this key has changed since
-        the last save.
-
-        """
-        if self._prev_dict is None:
-            return True
-        return self.previous(key) != self.get(key)
-
-    def previous(self, key):
-        """Return previous value for this key, or None if there
-        is no "previous" value.
-
-        """
-        if self._prev_dict:
-            return self._prev_dict.get(key)
-        return None
-
-    def save(self):
-        """Save this config to disk.
-
-        Preserves items in _prev_dict that do not exist in self.
-
-        """
-        if self._prev_dict:
-            for k, v in self._prev_dict.iteritems():
-                if k not in self:
-                    self[k] = v
-        with open(self.path, 'w') as f:
-            json.dump(self, f)
-
-
-@cached
-def config(scope=None):
-    """Juju charm configuration"""
-    config_cmd_line = ['config-get']
-    if scope is not None:
-        config_cmd_line.append(scope)
-    config_cmd_line.append('--format=json')
-    try:
-        config_data = json.loads(subprocess.check_output(config_cmd_line))
-        if scope is not None:
-            return config_data
-        return Config(config_data)
-    except ValueError:
-        return None
-
-
-@cached
-def relation_get(attribute=None, unit=None, rid=None):
-    """Get relation information"""
-    _args = ['relation-get', '--format=json']
-    if rid:
-        _args.append('-r')
-        _args.append(rid)
-    _args.append(attribute or '-')
-    if unit:
-        _args.append(unit)
-    try:
-        return json.loads(subprocess.check_output(_args))
-    except ValueError:
-        return None
-    except CalledProcessError, e:
-        if e.returncode == 2:
-            return None
-        raise
-
-
-def relation_set(relation_id=None, relation_settings={}, **kwargs):
-    """Set relation information for the current unit"""
-    relation_cmd_line = ['relation-set']
-    if relation_id is not None:
-        relation_cmd_line.extend(('-r', relation_id))
-    for k, v in (relation_settings.items() + kwargs.items()):
-        if v is None:
-            relation_cmd_line.append('{}='.format(k))
-        else:
-            relation_cmd_line.append('{}={}'.format(k, v))
-    subprocess.check_call(relation_cmd_line)
-    # Flush cache of any relation-gets for local unit
-    flush(local_unit())
-
-
-@cached
-def relation_ids(reltype=None):
-    """A list of relation_ids"""
-    reltype = reltype or relation_type()
-    relid_cmd_line = ['relation-ids', '--format=json']
-    if reltype is not None:
-        relid_cmd_line.append(reltype)
-        return json.loads(subprocess.check_output(relid_cmd_line)) or []
-    return []
-
-
-@cached
-def related_units(relid=None):
-    """A list of related units"""
-    relid = relid or relation_id()
-    units_cmd_line = ['relation-list', '--format=json']
-    if relid is not None:
-        units_cmd_line.extend(('-r', relid))
-    return json.loads(subprocess.check_output(units_cmd_line)) or []
-
-
-@cached
-def relation_for_unit(unit=None, rid=None):
-    """Get the json represenation of a unit's relation"""
-    unit = unit or remote_unit()
-    relation = relation_get(unit=unit, rid=rid)
-    for key in relation:
-        if key.endswith('-list'):
-            relation[key] = relation[key].split()
-    relation['__unit__'] = unit
-    return relation
-
-
-@cached
-def relations_for_id(relid=None):
-    """Get relations of a specific relation ID"""
-    relation_data = []
-    relid = relid or relation_ids()
-    for unit in related_units(relid):
-        unit_data = relation_for_unit(unit, relid)
-        unit_data['__relid__'] = relid
-        relation_data.append(unit_data)
-    return relation_data
-
-
-@cached
-def relations_of_type(reltype=None):
-    """Get relations of a specific type"""
-    relation_data = []
-    reltype = reltype or relation_type()
-    for relid in relation_ids(reltype):
-        for relation in relations_for_id(relid):
-            relation['__relid__'] = relid
-            relation_data.append(relation)
-    return relation_data
-
-
-@cached
-def relation_types():
-    """Get a list of relation types supported by this charm"""
-    charmdir = os.environ.get('CHARM_DIR', '')
-    mdf = open(os.path.join(charmdir, 'metadata.yaml'))
-    md = yaml.safe_load(mdf)
-    rel_types = []
-    for key in ('provides', 'requires', 'peers'):
-        section = md.get(key)
-        if section:
-            rel_types.extend(section.keys())
-    mdf.close()
-    return rel_types
-
-
-@cached
-def relations():
-    """Get a nested dictionary of relation data for all related units"""
-    rels = {}
-    for reltype in relation_types():
-        relids = {}
-        for relid in relation_ids(reltype):
-            units = {local_unit(): relation_get(unit=local_unit(), rid=relid)}
-            for unit in related_units(relid):
-                reldata = relation_get(unit=unit, rid=relid)
-                units[unit] = reldata
-            relids[relid] = units
-        rels[reltype] = relids
-    return rels
-
-
-@cached
-def is_relation_made(relation, keys='private-address'):
-    '''
-    Determine whether a relation is established by checking for
-    presence of key(s).  If a list of keys is provided, they
-    must all be present for the relation to be identified as made
-    '''
-    if isinstance(keys, str):
-        keys = [keys]
-    for r_id in relation_ids(relation):
-        for unit in related_units(r_id):
-            context = {}
-            for k in keys:
-                context[k] = relation_get(k, rid=r_id,
-                                          unit=unit)
-            if None not in context.values():
-                return True
-    return False
-
-
-def open_port(port, protocol="TCP"):
-    """Open a service network port"""
-    _args = ['open-port']
-    _args.append('{}/{}'.format(port, protocol))
-    subprocess.check_call(_args)
-
-
-def close_port(port, protocol="TCP"):
-    """Close a service network port"""
-    _args = ['close-port']
-    _args.append('{}/{}'.format(port, protocol))
-    subprocess.check_call(_args)
-
-
-@cached
-def unit_get(attribute):
-    """Get the unit ID for the remote unit"""
-    _args = ['unit-get', '--format=json', attribute]
-    try:
-        return json.loads(subprocess.check_output(_args))
-    except ValueError:
-        return None
-
-
-def unit_private_ip():
-    """Get this unit's private IP address"""
-    return unit_get('private-address')
-
-
-class UnregisteredHookError(Exception):
-    """Raised when an undefined hook is called"""
-    pass
-
-
-class Hooks(object):
-    """A convenient handler for hook functions.
-
-    Example:
-        hooks = Hooks()
-
-        # register a hook, taking its name from the function name
-        @hooks.hook()
-        def install():
-            ...
-
-        # register a hook, providing a custom hook name
-        @hooks.hook("config-changed")
-        def config_changed():
-            ...
-
-        if __name__ == "__main__":
-            # execute a hook based on the name the program is called by
-            hooks.execute(sys.argv)
-    """
-
-    def __init__(self):
-        super(Hooks, self).__init__()
-        self._hooks = {}
-
-    def register(self, name, function):
-        """Register a hook"""
-        self._hooks[name] = function
-
-    def execute(self, args):
-        """Execute a registered hook based on args[0]"""
-        hook_name = os.path.basename(args[0])
-        if hook_name in self._hooks:
-            self._hooks[hook_name]()
-        else:
-            raise UnregisteredHookError(hook_name)
-
-    def hook(self, *hook_names):
-        """Decorator, registering them as hooks"""
-        def wrapper(decorated):
-            for hook_name in hook_names:
-                self.register(hook_name, decorated)
-            else:
-                self.register(decorated.__name__, decorated)
-                if '_' in decorated.__name__:
-                    self.register(
-                        decorated.__name__.replace('_', '-'), decorated)
-            return decorated
-        return wrapper
-
-
-def charm_dir():
-    """Return the root directory of the current charm"""
-    return os.environ.get('CHARM_DIR')
 
=== removed file 'hooks/charmhelpers/core/host.py'
--- hooks/charmhelpers/core/host.py	2014-07-09 12:37:36 +0000
+++ hooks/charmhelpers/core/host.py	1970-01-01 00:00:00 +0000
@@ -1,325 +0,0 @@
-"""Tools for working with the host system"""
-# Copyright 2012 Canonical Ltd.
-#
-# Authors:
-#  Nick Moffitt <[e-mail]>
-#  Matthew Wedgwood <[e-mail]>
-
-import os
-import pwd
-import grp
-import random
-import string
-import subprocess
-import hashlib
-import apt_pkg
-
-from collections import OrderedDict
-
-from hookenv import log
-from fstab import Fstab
-
-
-def service_start(service_name):
-    """Start a system service"""
-    return service('start', service_name)
-
-
-def service_stop(service_name):
-    """Stop a system service"""
-    return service('stop', service_name)
-
-
-def service_restart(service_name):
-    """Restart a system service"""
-    return service('restart', service_name)
-
-
-def service_reload(service_name, restart_on_failure=False):
-    """Reload a system service, optionally falling back to restart if
-    reload fails"""
-    service_result = service('reload', service_name)
-    if not service_result and restart_on_failure:
-        service_result = service('restart', service_name)
-    return service_result
-
-
-def service(action, service_name):
-    """Control a system service"""
-    cmd = ['service', service_name, action]
-    return subprocess.call(cmd) == 0
-
-
-def service_running(service):
-    """Determine whether a system service is running"""
-    try:
-        output = subprocess.check_output(['service', service, 'status'])
-    except subprocess.CalledProcessError:
-        return False
-    else:
-        if ("start/running" in output or "is running" in output):
-            return True
-        else:
-            return False
-
-
-def adduser(username, password=None, shell='/bin/bash', system_user=False):
-    """Add a user to the system"""
-    try:
-        user_info = pwd.getpwnam(username)
-        log('user {0} already exists!'.format(username))
-    except KeyError:
-        log('creating user {0}'.format(username))
-        cmd = ['useradd']
-        if system_user or password is None:
-            cmd.append('--system')
-        else:
-            cmd.extend([
-                '--create-home',
-                '--shell', shell,
-                '--password', password,
-            ])
-        cmd.append(username)
-        subprocess.check_call(cmd)
-        user_info = pwd.getpwnam(username)
-    return user_info
-
-
-def add_user_to_group(username, group):
-    """Add a user to a group"""
-    cmd = [
-        'gpasswd', '-a',
-        username,
-        group
-    ]
-    log("Adding user {} to group {}".format(username, group))
-    subprocess.check_call(cmd)
-
-
-def rsync(from_path, to_path, flags='-r', options=None):
-    """Replicate the contents of a path"""
-    options = options or ['--delete', '--executability']
-    cmd = ['/usr/bin/rsync', flags]
-    cmd.extend(options)
-    cmd.append(from_path)
-    cmd.append(to_path)
-    log(" ".join(cmd))
-    return subprocess.check_output(cmd).strip()
-
-
-def symlink(source, destination):
-    """Create a symbolic link"""
-    log("Symlinking {} as {}".format(source, destination))
-    cmd = [
-        'ln',
-        '-sf',
-        source,
-        destination,
-    ]
-    subprocess.check_call(cmd)
-
-
-def mkdir(path, owner='root', group='root', perms=0555, force=False):
-    """Create a directory"""
-    log("Making dir {} {}:{} {:o}".format(path, owner, group,
-                                          perms))
-    uid = pwd.getpwnam(owner).pw_uid
-    gid = grp.getgrnam(group).gr_gid
-    realpath = os.path.abspath(path)
-    if os.path.exists(realpath):
-        if force and not os.path.isdir(realpath):
-            log("Removing non-directory file {} prior to mkdir()".format(path))
-            os.unlink(realpath)
-    else:
-        os.makedirs(realpath, perms)
-    os.chown(realpath, uid, gid)
-
-
-def write_file(path, content, owner='root', group='root', perms=0444):
-    """Create or overwrite a file with the contents of a string"""
-    log("Writing file {} {}:{} {:o}".format(path, owner, group, perms))
-    uid = pwd.getpwnam(owner).pw_uid
-    gid = grp.getgrnam(group).gr_gid
-    with open(path, 'w') as target:
-        os.fchown(target.fileno(), uid, gid)
-        os.fchmod(target.fileno(), perms)
-        target.write(content)
-
-
-def fstab_remove(mp):
-    """Remove the given mountpoint entry from /etc/fstab
-    """
-    return Fstab.remove_by_mountpoint(mp)
-
-
-def fstab_add(dev, mp, fs, options=None):
-    """Adds the given device entry to the /etc/fstab file
-    """
-    return Fstab.add(dev, mp, fs, options=options)
-
-
-def mount(device, mountpoint, options=None, persist=False, filesystem="ext3"):
-    """Mount a filesystem at a particular mountpoint"""
-    cmd_args = ['mount']
-    if options is not None:
-        cmd_args.extend(['-o', options])
-    cmd_args.extend([device, mountpoint])
-    try:
-        subprocess.check_output(cmd_args)
-    except subprocess.CalledProcessError, e:
-        log('Error mounting {} at {}\n{}'.format(device, mountpoint, e.output))
-        return False
-
-    if persist:
-        return fstab_add(device, mountpoint, filesystem, options=options)
-    return True
-
-
-def umount(mountpoint, persist=False):
-    """Unmount a filesystem"""
-    cmd_args = ['umount', mountpoint]
-    try:
-        subprocess.check_output(cmd_args)
-    except subprocess.CalledProcessError, e:
-        log('Error unmounting {}\n{}'.format(mountpoint, e.output))
-        return False
-
-    if persist:
-        return fstab_remove(mountpoint)
-    return True
-
-
-def mounts():
-    """Get a list of all mounted volumes as [[mountpoint,device],[...]]"""
-    with open('/proc/mounts') as f:
-        # [['/mount/point','/dev/path'],[...]]
-        system_mounts = [m[1::-1] for m in [l.strip().split()
-                                            for l in f.readlines()]]
-    return system_mounts
-
-
-def file_hash(path):
-    """Generate a md5 hash of the contents of 'path' or None if not found """
-    if os.path.exists(path):
-        h = hashlib.md5()
-        with open(path, 'r') as source:
-            h.update(source.read())  # IGNORE:E1101 - it does have update
-        return h.hexdigest()
-    else:
-        return None
-
-
-def restart_on_change(restart_map, stopstart=False):
-    """Restart services based on configuration files changing
-
-    This function is used a decorator, for example
-
-        @restart_on_change({
-            '/etc/ceph/ceph.conf': [ 'cinder-api', 'cinder-volume' ]
-            })
-        def ceph_client_changed():
-            ...
-
-    In this example, the cinder-api and cinder-volume services
-    would be restarted if /etc/ceph/ceph.conf is changed by the
-    ceph_client_changed function.
-    """
-    def wrap(f):
-        def wrapped_f(*args):
-            checksums = {}
-            for path in restart_map:
-                checksums[path] = file_hash(path)
-            f(*args)
-            restarts = []
-            for path in restart_map:
-                if checksums[path] != file_hash(path):
-                    restarts += restart_map[path]
-            services_list = list(OrderedDict.fromkeys(restarts))
-            if not stopstart:
-                for service_name in services_list:
-                    service('restart', service_name)
-            else:
-                for action in ['stop', 'start']:
-                    for service_name in services_list:
-                        service(action, service_name)
-        return wrapped_f
-    return wrap
-
-
-def lsb_release():
-    """Return /etc/lsb-release in a dict"""
-    d = {}
-    with open('/etc/lsb-release', 'r') as lsb:
-        for l in lsb:
-            k, v = l.split('=')
-            d[k.strip()] = v.strip()
-    return d
-
-
-def pwgen(length=None):
-    """Generate a random pasword."""
-    if length is None:
-        length = random.choice(range(35, 45))
-    alphanumeric_chars = [
-        l for l in (string.letters + string.digits)
-        if l not in 'l0QD1vAEIOUaeiou']
-    random_chars = [
-        random.choice(alphanumeric_chars) for _ in range(length)]
-    return(''.join(random_chars))
-
-
-def list_nics(nic_type):
-    '''Return a list of nics of given type(s)'''
-    if isinstance(nic_type, basestring):
-        int_types = [nic_type]
-    else:
-        int_types = nic_type
-    interfaces = []
-    for int_type in int_types:
-        cmd = ['ip', 'addr', 'show', 'label', int_type + '*']
-        ip_output = subprocess.check_output(cmd).split('\n')
-        ip_output = (line for line in ip_output if line)
-        for line in ip_output:
-            if line.split()[1].startswith(int_type):
-                interfaces.append(line.split()[1].replace(":", ""))
-    return interfaces
-
-
-def set_nic_mtu(nic, mtu):
-    '''Set MTU on a network interface'''
-    cmd = ['ip', 'link', 'set', nic, 'mtu', mtu]
-    subprocess.check_call(cmd)
-
-
-def get_nic_mtu(nic):
-    cmd = ['ip', 'addr', 'show', nic]
-    ip_output = subprocess.check_output(cmd).split('\n')
-    mtu = ""
-    for line in ip_output:
-        words = line.split()
-        if 'mtu' in words:
-            mtu = words[words.index("mtu") + 1]
-    return mtu
-
-
-def get_nic_hwaddr(nic):
-    cmd = ['ip', '-o', '-0', 'addr', 'show', nic]
-    ip_output = subprocess.check_output(cmd)
-    hwaddr = ""
-    words = ip_output.split()
-    if 'link/ether' in words:
-        hwaddr = words[words.index('link/ether') + 1]
-    return hwaddr
-
-
-def cmp_pkgrevno(package, revno, pkgcache=None):
-    '''Compare supplied revno with the revno of the installed package
-       1 => Installed revno is greater than supplied arg
-       0 => Installed revno is the same as supplied arg
-      -1 => Installed revno is less than supplied arg
-    '''
-    if not pkgcache:
-        apt_pkg.init()
-        pkgcache = apt_pkg.Cache()
-    pkg = pkgcache[package]
-    return apt_pkg.version_compare(pkg.current_ver.ver_str, revno)
 
=== removed directory 'hooks/charmhelpers/fetch'
=== removed file 'hooks/charmhelpers/fetch/__init__.py'
--- hooks/charmhelpers/fetch/__init__.py	2014-07-09 12:37:36 +0000
+++ hooks/charmhelpers/fetch/__init__.py	1970-01-01 00:00:00 +0000
@@ -1,349 +0,0 @@
-import importlib
-import time
-from yaml import safe_load
-from charmhelpers.core.host import (
-    lsb_release
-)
-from urlparse import (
-    urlparse,
-    urlunparse,
-)
-import subprocess
-from charmhelpers.core.hookenv import (
-    config,
-    log,
-)
-import apt_pkg
-import os
-
-
-CLOUD_ARCHIVE = """# Ubuntu Cloud Archive
-deb http://ubuntu-cloud.archive.canonical.com/ubuntu {} main
-"""
-PROPOSED_POCKET = """# Proposed
-deb http://archive.ubuntu.com/ubuntu {}-proposed main universe multiverse restricted
-"""
-CLOUD_ARCHIVE_POCKETS = {
-    # Folsom
-    'folsom': 'precise-updates/folsom',
-    'precise-folsom': 'precise-updates/folsom',
-    'precise-folsom/updates': 'precise-updates/folsom',
-    'precise-updates/folsom': 'precise-updates/folsom',
-    'folsom/proposed': 'precise-proposed/folsom',
-    'precise-folsom/proposed': 'precise-proposed/folsom',
-    'precise-proposed/folsom': 'precise-proposed/folsom',
-    # Grizzly
-    'grizzly': 'precise-updates/grizzly',
-    'precise-grizzly': 'precise-updates/grizzly',
-    'precise-grizzly/updates': 'precise-updates/grizzly',
-    'precise-updates/grizzly': 'precise-updates/grizzly',
-    'grizzly/proposed': 'precise-proposed/grizzly',
-    'precise-grizzly/proposed': 'precise-proposed/grizzly',
-    'precise-proposed/grizzly': 'precise-proposed/grizzly',
-    # Havana
-    'havana': 'precise-updates/havana',
-    'precise-havana': 'precise-updates/havana',
-    'precise-havana/updates': 'precise-updates/havana',
-    'precise-updates/havana': 'precise-updates/havana',
-    'havana/proposed': 'precise-proposed/havana',
-    'precise-havana/proposed': 'precise-proposed/havana',
-    'precise-proposed/havana': 'precise-proposed/havana',
-    # Icehouse
-    'icehouse': 'precise-updates/icehouse',
-    'precise-icehouse': 'precise-updates/icehouse',
-    'precise-icehouse/updates': 'precise-updates/icehouse',
-    'precise-updates/icehouse': 'precise-updates/icehouse',
-    'icehouse/proposed': 'precise-proposed/icehouse',
-    'precise-icehouse/proposed': 'precise-proposed/icehouse',
-    'precise-proposed/icehouse': 'precise-proposed/icehouse',
-    # Juno
-    'juno': 'trusty-updates/juno',
-    'trusty-juno': 'trusty-updates/juno',
-    'trusty-juno/updates': 'trusty-updates/juno',
-    'trusty-updates/juno': 'trusty-updates/juno',
-    'juno/proposed': 'trusty-proposed/juno',
-    'juno/proposed': 'trusty-proposed/juno',
-    'trusty-juno/proposed': 'trusty-proposed/juno',
-    'trusty-proposed/juno': 'trusty-proposed/juno',
-}
-
-# The order of this list is very important. Handlers should be listed in from
-# least- to most-specific URL matching.
-FETCH_HANDLERS = (
-    'charmhelpers.fetch.archiveurl.ArchiveUrlFetchHandler',
-    'charmhelpers.fetch.bzrurl.BzrUrlFetchHandler',
-)
-
-APT_NO_LOCK = 100  # The return code for "couldn't acquire lock" in APT.
-APT_NO_LOCK_RETRY_DELAY = 10  # Wait 10 seconds between apt lock checks.
-APT_NO_LOCK_RETRY_COUNT = 30  # Retry to acquire the lock X times.
-
-
-class SourceConfigError(Exception):
-    pass
-
-
-class UnhandledSource(Exception):
-    pass
-
-
-class AptLockError(Exception):
-    pass
-
-
-class BaseFetchHandler(object):
-
-    """Base class for FetchHandler implementations in fetch plugins"""
-
-    def can_handle(self, source):
-        """Returns True if the source can be handled. Otherwise returns
-        a string explaining why it cannot"""
-        return "Wrong source type"
-
-    def install(self, source):
-        """Try to download and unpack the source. Return the path to the
-        unpacked files or raise UnhandledSource."""
-        raise UnhandledSource("Wrong source type {}".format(source))
-
-    def parse_url(self, url):
-        return urlparse(url)
-
-    def base_url(self, url):
-        """Return url without querystring or fragment"""
-        parts = list(self.parse_url(url))
-        parts[4:] = ['' for i in parts[4:]]
-        return urlunparse(parts)
-
-
-def filter_installed_packages(packages):
-    """Returns a list of packages that require installation"""
-    apt_pkg.init()
-
-    # Tell apt to build an in-memory cache to prevent race conditions (if
-    # another process is already building the cache).
-    apt_pkg.config.set("Dir::Cache::pkgcache", "")
-
-    cache = apt_pkg.Cache()
-    _pkgs = []
-    for package in packages:
-        try:
-            p = cache[package]
-            p.current_ver or _pkgs.append(package)
-        except KeyError:
-            log('Package {} has no installation candidate.'.format(package),
-                level='WARNING')
-            _pkgs.append(package)
-    return _pkgs
-
-
-def apt_install(packages, options=None, fatal=False):
-    """Install one or more packages"""
-    if options is None:
-        options = ['--option=Dpkg::Options::=--force-confold']
-
-    cmd = ['apt-get', '--assume-yes']
-    cmd.extend(options)
-    cmd.append('install')
-    if isinstance(packages, basestring):
-        cmd.append(packages)
-    else:
-        cmd.extend(packages)
-    log("Installing {} with options: {}".format(packages,
-                                                options))
-    _run_apt_command(cmd, fatal)
-
-
-def apt_upgrade(options=None, fatal=False, dist=False):
-    """Upgrade all packages"""
-    if options is None:
-        options = ['--option=Dpkg::Options::=--force-confold']
-
-    cmd = ['apt-get', '--assume-yes']
-    cmd.extend(options)
-    if dist:
-        cmd.append('dist-upgrade')
-    else:
-        cmd.append('upgrade')
-    log("Upgrading with options: {}".format(options))
-    _run_apt_command(cmd, fatal)
-
-
-def apt_update(fatal=False):
-    """Update local apt cache"""
-    cmd = ['apt-get', 'update']
-    _run_apt_command(cmd, fatal)
-
-
-def apt_purge(packages, fatal=False):
-    """Purge one or more packages"""
-    cmd = ['apt-get', '--assume-yes', 'purge']
-    if isinstance(packages, basestring):
-        cmd.append(packages)
-    else:
-        cmd.extend(packages)
-    log("Purging {}".format(packages))
-    _run_apt_command(cmd, fatal)
-
-
-def apt_hold(packages, fatal=False):
-    """Hold one or more packages"""
-    cmd = ['apt-mark', 'hold']
-    if isinstance(packages, basestring):
-        cmd.append(packages)
-    else:
-        cmd.extend(packages)
-    log("Holding {}".format(packages))
-
-    if fatal:
-        subprocess.check_call(cmd)
-    else:
-        subprocess.call(cmd)
-
-
-def add_source(source, key=None):
-    if source is None:
-        log('Source is not present. Skipping')
-        return
-
-    if (source.startswith('ppa:') or
-        source.startswith('http') or
-        source.startswith('deb ') or
-            source.startswith('cloud-archive:')):
-        subprocess.check_call(['add-apt-repository', '--yes', source])
-    elif source.startswith('cloud:'):
-        apt_install(filter_installed_packages(['ubuntu-cloud-keyring']),
-                    fatal=True)
-        pocket = source.split(':')[-1]
-        if pocket not in CLOUD_ARCHIVE_POCKETS:
-            raise SourceConfigError(
-                'Unsupported cloud: source option %s' %
-                pocket)
-        actual_pocket = CLOUD_ARCHIVE_POCKETS[pocket]
-        with open('/etc/apt/sources.list.d/cloud-archive.list', 'w') as apt:
-            apt.write(CLOUD_ARCHIVE.format(actual_pocket))
-    elif source == 'proposed':
-        release = lsb_release()['DISTRIB_CODENAME']
-        with open('/etc/apt/sources.list.d/proposed.list', 'w') as apt:
-            apt.write(PROPOSED_POCKET.format(release))
-    if key:
-        subprocess.check_call(['apt-key', 'adv', '--keyserver',
-                               'hkp://keyserver.ubuntu.com:80', '--recv',
-                               key])
-
-
-def configure_sources(update=False,
-                      sources_var='install_sources',
-                      keys_var='install_keys'):
-    """
-    Configure multiple sources from charm configuration
-
-    Example config:
-        install_sources:
-          - "ppa:foo"
-          - "http://example.com/repo precise main"
-        install_keys:
-          - null
-          - "a1b2c3d4"
-
-    Note that 'null' (a.k.a. None) should not be quoted.
-    """
-    sources = safe_load(config(sources_var))
-    keys = config(keys_var)
-    if keys is not None:
-        keys = safe_load(keys)
-    if isinstance(sources, basestring) and (
-            keys is None or isinstance(keys, basestring)):
-        add_source(sources, keys)
-    else:
-        if not len(sources) == len(keys):
-            msg = 'Install sources and keys lists are different lengths'
-            raise SourceConfigError(msg)
-        for src_num in range(len(sources)):
-            add_source(sources[src_num], keys[src_num])
-    if update:
-        apt_update(fatal=True)
-
-
-def install_remote(source):
-    """
-    Install a file tree from a remote source
-
-    The specified source should be a url of the form:
-        scheme://[host]/path[#[option=value][&...]]
-
-    Schemes supported are based on this modules submodules
-    Options supported are submodule-specific"""
-    # We ONLY check for True here because can_handle may return a string
-    # explaining why it can't handle a given source.
-    handlers = [h for h in plugins() if h.can_handle(source) is True]
-    installed_to = None
-    for handler in handlers:
-        try:
-            installed_to = handler.install(source)
-        except UnhandledSource:
-            pass
-    if not installed_to:
-        raise UnhandledSource("No handler found for source {}".format(source))
-    return installed_to
-
-
-def install_from_config(config_var_name):
-    charm_config = config()
-    source = charm_config[config_var_name]
-    return install_remote(source)
-
-
-def plugins(fetch_handlers=None):
-    if not fetch_handlers:
-        fetch_handlers = FETCH_HANDLERS
-    plugin_list = []
-    for handler_name in fetch_handlers:
-        package, classname = handler_name.rsplit('.', 1)
-        try:
-            handler_class = getattr(
-                importlib.import_module(package),
-                classname)
-            plugin_list.append(handler_class())
-        except (ImportError, AttributeError):
-            # Skip missing plugins so that they can be ommitted from
-            # installation if desired
-            log("FetchHandler {} not found, skipping plugin".format(
-                handler_name))
-    return plugin_list
-
-
-def _run_apt_command(cmd, fatal=False):
-    """
-    Run an APT command, checking output and retrying if the fatal flag is set
-    to True.
-
-    :param: cmd: str: The apt command to run.
-    :param: fatal: bool: Whether the command's output should be checked and
-        retried.
-    """
-    env = os.environ.copy()
-
-    if 'DEBIAN_FRONTEND' not in env:
-        env['DEBIAN_FRONTEND'] = 'noninteractive'
-
-    if fatal:
-        retry_count = 0
-        result = None
-
-        # If the command is considered "fatal", we need to retry if the apt
-        # lock was not acquired.
-
-        while result is None or result == APT_NO_LOCK:
-            try:
-                result = subprocess.check_call(cmd, env=env)
-            except subprocess.CalledProcessError, e:
-                retry_count = retry_count + 1
-                if retry_count > APT_NO_LOCK_RETRY_COUNT:
-                    raise
-                result = e.returncode
-                log("Couldn't acquire DPKG lock. Will retry in {} seconds."
-                    "".format(APT_NO_LOCK_RETRY_DELAY))
-                time.sleep(APT_NO_LOCK_RETRY_DELAY)
-
-    else:
-        subprocess.call(cmd, env=env)
 
=== removed file 'hooks/charmhelpers/fetch/archiveurl.py'
--- hooks/charmhelpers/fetch/archiveurl.py	2014-07-09 12:37:36 +0000
+++ hooks/charmhelpers/fetch/archiveurl.py	1970-01-01 00:00:00 +0000
@@ -1,63 +0,0 @@
-import os
-import urllib2
-import urlparse
-
-from charmhelpers.fetch import (
-    BaseFetchHandler,
-    UnhandledSource
-)
-from charmhelpers.payload.archive import (
-    get_archive_handler,
-    extract,
-)
-from charmhelpers.core.host import mkdir
-
-
-class ArchiveUrlFetchHandler(BaseFetchHandler):
-    """Handler for archives via generic URLs"""
-    def can_handle(self, source):
-        url_parts = self.parse_url(source)
-        if url_parts.scheme not in ('http', 'https', 'ftp', 'file'):
-            return "Wrong source type"
-        if get_archive_handler(self.base_url(source)):
-            return True
-        return False
-
-    def download(self, source, dest):
-        # propogate all exceptions
-        # URLError, OSError, etc
-        proto, netloc, path, params, query, fragment = urlparse.urlparse(source)
-        if proto in ('http', 'https'):
-            auth, barehost = urllib2.splituser(netloc)
-            if auth is not None:
-                source = urlparse.urlunparse((proto, barehost, path, params, query, fragment))
-                username, password = urllib2.splitpasswd(auth)
-                passman = urllib2.HTTPPasswordMgrWithDefaultRealm()
-                # Realm is set to None in add_password to force the username and password
-                # to be used whatever the realm
-                passman.add_password(None, source, username, password)
-                authhandler = urllib2.HTTPBasicAuthHandler(passman)
-                opener = urllib2.build_opener(authhandler)
-                urllib2.install_opener(opener)
-        response = urllib2.urlopen(source)
-        try:
-            with open(dest, 'w') as dest_file:
-                dest_file.write(response.read())
-        except Exception as e:
-            if os.path.isfile(dest):
-                os.unlink(dest)
-            raise e
-
-    def install(self, source):
-        url_parts = self.parse_url(source)
-        dest_dir = os.path.join(os.environ.get('CHARM_DIR'), 'fetched')
-        if not os.path.exists(dest_dir):
-            mkdir(dest_dir, perms=0755)
-        dld_file = os.path.join(dest_dir, os.path.basename(url_parts.path))
-        try:
-            self.download(source, dld_file)
-        except urllib2.URLError as e:
-            raise UnhandledSource(e.reason)
-        except OSError as e:
-            raise UnhandledSource(e.strerror)
-        return extract(dld_file)
 
=== removed file 'hooks/charmhelpers/fetch/bzrurl.py'
--- hooks/charmhelpers/fetch/bzrurl.py	2014-07-09 12:37:36 +0000
+++ hooks/charmhelpers/fetch/bzrurl.py	1970-01-01 00:00:00 +0000
@@ -1,50 +0,0 @@
-import os
-from charmhelpers.fetch import (
-    BaseFetchHandler,
-    UnhandledSource
-)
-from charmhelpers.core.host import mkdir
-
-try:
-    from bzrlib.branch import Branch
-except ImportError:
-    from charmhelpers.fetch import apt_install
-    apt_install("python-bzrlib")
-    from bzrlib.branch import Branch
-
-
-class BzrUrlFetchHandler(BaseFetchHandler):
-    """Handler for bazaar branches via generic and lp URLs"""
-    def can_handle(self, source):
-        url_parts = self.parse_url(source)
-        if url_parts.scheme not in ('bzr+ssh', 'lp'):
-            return False
-        else:
-            return True
-
-    def branch(self, source, dest):
-        url_parts = self.parse_url(source)
-        # If we use lp:branchname scheme we need to load plugins
-        if not self.can_handle(source):
-            raise UnhandledSource("Cannot handle {}".format(source))
-        if url_parts.scheme == "lp":
-            from bzrlib.plugin import load_plugins
-            load_plugins()
-        try:
-            remote_branch = Branch.open(source)
-            remote_branch.bzrdir.sprout(dest).open_branch()
-        except Exception as e:
-            raise e
-
-    def install(self, source):
-        url_parts = self.parse_url(source)
-        branch_name = url_parts.path.strip("/").split("/")[-1]
-        dest_dir = os.path.join(os.environ.get('CHARM_DIR'), "fetched",
-                                branch_name)
-        if not os.path.exists(dest_dir):
-            mkdir(dest_dir, perms=0755)
-        try:
-            self.branch(source, dest_dir)
-        except OSError as e:
-            raise UnhandledSource(e.strerror)
-        return dest_dir
 
=== removed directory 'hooks/charmhelpers/lib'
=== removed file 'hooks/charmhelpers/lib/__init__.py'
=== removed file 'hooks/charmhelpers/lib/ceph_utils.py'
--- hooks/charmhelpers/lib/ceph_utils.py	2014-07-09 12:37:36 +0000
+++ hooks/charmhelpers/lib/ceph_utils.py	1970-01-01 00:00:00 +0000
@@ -1,315 +0,0 @@
-#
-# Copyright 2012 Canonical Ltd.
-#
-# This file is sourced from lp:openstack-charm-helpers
-#
-# Authors:
-#  James Page <[e-mail]>
-#  Adam Gandelman <[e-mail]>
-#
-
-import commands
-import json
-import subprocess
-import os
-import shutil
-import time
-import lib.utils as utils
-
-KEYRING = '/etc/ceph/ceph.client.%s.keyring'
-KEYFILE = '/etc/ceph/ceph.client.%s.key'
-
-CEPH_CONF = """[global]
- auth supported = %(auth)s
- keyring = %(keyring)s
- mon host = %(mon_hosts)s
- log to syslog = %(use_syslog)s
- err to syslog = %(use_syslog)s
- clog to syslog = %(use_syslog)s
-"""
-
-
-def execute(cmd):
-    subprocess.check_call(cmd)
-
-
-def execute_shell(cmd):
-    subprocess.check_call(cmd, shell=True)
-
-
-def install():
-    ceph_dir = "/etc/ceph"
-    if not os.path.isdir(ceph_dir):
-        os.mkdir(ceph_dir)
-    utils.install('ceph-common')
-
-
-def rbd_exists(service, pool, rbd_img):
-    (rc, out) = commands.getstatusoutput('rbd list --id %s --pool %s' %
-                                         (service, pool))
-    return rbd_img in out
-
-
-def create_rbd_image(service, pool, image, sizemb):
-    cmd = [
-        'rbd',
-        'create',
-        image,
-        '--size',
-        str(sizemb),
-        '--id',
-        service,
-        '--pool',
-        pool]
-    execute(cmd)
-
-
-def pool_exists(service, name):
-    (rc, out) = commands.getstatusoutput("rados --id %s lspools" % service)
-    return name in out
-
-
-def ceph_version():
-    ''' Retrieve the local version of ceph '''
-    if os.path.exists('/usr/bin/ceph'):
-        cmd = ['ceph', '-v']
-        output = subprocess.check_output(cmd)
-        output = output.split()
-        if len(output) > 3:
-            return output[2]
-        else:
-            return None
-    else:
-        return None
-
-
-def get_osds(service):
-    '''
-    Return a list of all Ceph Object Storage Daemons
-    currently in the cluster
-    '''
-    version = ceph_version()
-    if version and version >= '0.56':
-        cmd = ['ceph', '--id', service, 'osd', 'ls', '--format=json']
-        return json.loads(subprocess.check_output(cmd))
-    else:
-        return None
-
-
-def create_pool(service, name, replicas=2):
-    ''' Create a new RADOS pool '''
-    if pool_exists(service, name):
-        utils.juju_log('WARNING',
-                       "Ceph pool {} already exists, "
-                       "skipping creation".format(name))
-        return
-
-    osds = get_osds(service)
-    if osds:
-        pgnum = (len(osds) * 100 / replicas)
-    else:
-        # NOTE(james-page): Default to 200 for older ceph versions
-        # which don't support OSD query from cli
-        pgnum = 200
-
-    cmd = [
-        'ceph', '--id', service,
-        'osd', 'pool', 'create',
-        name, str(pgnum)
-    ]
-    subprocess.check_call(cmd)
-    cmd = [
-        'ceph', '--id', service,
-        'osd', 'pool', 'set', name,
-        'size', str(replicas)
-    ]
-    subprocess.check_call(cmd)
-
-
-def keyfile_path(service):
-    return KEYFILE % service
-
-
-def keyring_path(service):
-    return KEYRING % service
-
-
-def create_keyring(service, key):
-    keyring = keyring_path(service)
-    if os.path.exists(keyring):
-        utils.juju_log('INFO', 'ceph: Keyring exists at %s.' % keyring)
-    cmd = [
-        'ceph-authtool',
-        keyring,
-        '--create-keyring',
-        '--name=client.%s' % service,
-        '--add-key=%s' % key]
-    execute(cmd)
-    utils.juju_log('INFO', 'ceph: Created new ring at %s.' % keyring)
-
-
-def create_key_file(service, key):
-    # create a file containing the key
-    keyfile = keyfile_path(service)
-    if os.path.exists(keyfile):
-        utils.juju_log('INFO', 'ceph: Keyfile exists at %s.' % keyfile)
-    fd = open(keyfile, 'w')
-    fd.write(key)
-    fd.close()
-    utils.juju_log('INFO', 'ceph: Created new keyfile at %s.' % keyfile)
-
-
-def get_ceph_nodes():
-    hosts = []
-    for r_id in utils.relation_ids('ceph'):
-        for unit in utils.relation_list(r_id):
-            hosts.append(utils.relation_get('private-address',
-                                            unit=unit, rid=r_id))
-    return hosts
-
-
-def configure(service, key, auth, use_syslog):
-    create_keyring(service, key)
-    create_key_file(service, key)
-    hosts = get_ceph_nodes()
-    mon_hosts = ",".join(map(str, hosts))
-    keyring = keyring_path(service)
-    with open('/etc/ceph/ceph.conf', 'w') as ceph_conf:
-        ceph_conf.write(CEPH_CONF % locals())
-    modprobe_kernel_module('rbd')
-
-
-def image_mapped(image_name):
-    (rc, out) = commands.getstatusoutput('rbd showmapped')
-    return image_name in out
-
-
-def map_block_storage(service, pool, image):
-    cmd = [
-        'rbd',
-        'map',
-        '%s/%s' % (pool, image),
-        '--user',
-        service,
-        '--secret',
-        keyfile_path(service)]
-    execute(cmd)
-
-
-def filesystem_mounted(fs):
-    return subprocess.call(['grep', '-wqs', fs, '/proc/mounts']) == 0
-
-
-def make_filesystem(blk_device, fstype='ext4'):
-    count = 0
-    e_noent = os.errno.ENOENT
-    while not os.path.exists(blk_device):
-        if count >= 10:
-            utils.juju_log('ERROR', 'ceph: gave up waiting on block '
-                                    'device %s' % blk_device)
-            raise IOError(e_noent, os.strerror(e_noent), blk_device)
-        utils.juju_log('INFO', 'ceph: waiting for block device %s '
-                               'to appear' % blk_device)
-        count += 1
-        time.sleep(1)
-    else:
-        utils.juju_log('INFO', 'ceph: Formatting block device %s '
-                               'as filesystem %s.' % (blk_device, fstype))
-        execute(['mkfs', '-t', fstype, blk_device])
-
-
-def place_data_on_ceph(service, blk_device, data_src_dst, fstype='ext4'):
-    # mount block device into /mnt
-    cmd = ['mount', '-t', fstype, blk_device, '/mnt']
-    execute(cmd)
-
-    # copy data to /mnt
-    try:
-        copy_files(data_src_dst, '/mnt')
-    except:
-        pass
-
-    # umount block device
-    cmd = ['umount', '/mnt']
-    execute(cmd)
-
-    _dir = os.stat(data_src_dst)
-    uid = _dir.st_uid
-    gid = _dir.st_gid
-
-    # re-mount where the data should originally be
-    cmd = ['mount', '-t', fstype, blk_device, data_src_dst]
-    execute(cmd)
-
-    # ensure original ownership of new mount.
-    cmd = ['chown', '-R', '%s:%s' % (uid, gid), data_src_dst]
-    execute(cmd)
-
-
-# TODO: re-use
-def modprobe_kernel_module(module):
-    utils.juju_log('INFO', 'Loading kernel module')
-    cmd = ['modprobe', module]
-    execute(cmd)
-    cmd = 'echo %s >> /etc/modules' % module
-    execute_shell(cmd)
-
-
-def copy_files(src, dst, symlinks=False, ignore=None):
-    for item in os.listdir(src):
-        s = os.path.join(src, item)
-        d = os.path.join(dst, item)
-        if os.path.isdir(s):
-            shutil.copytree(s, d, symlinks, ignore)
-        else:
-            shutil.copy2(s, d)
-
-
-def ensure_ceph_storage(service, pool, rbd_img, sizemb, mount_point,
-                        blk_device, fstype, system_services=[],
-                        rbd_pool_replicas=2):
-    """
-    To be called from the current cluster leader.
-    Ensures given pool and RBD image exists, is mapped to a block device,
-    and the device is formatted and mounted at the given mount_point.
-
-    If formatting a device for the first time, data existing at mount_point
-    will be migrated to the RBD device before being remounted.
-
-    All services listed in system_services will be stopped prior to data
-    migration and restarted when complete.
-    """
-    # Ensure pool, RBD image, RBD mappings are in place.
-    if not pool_exists(service, pool):
-        utils.juju_log('INFO', 'ceph: Creating new pool %s.' % pool)
-        create_pool(service, pool, replicas=rbd_pool_replicas)
-
-    if not rbd_exists(service, pool, rbd_img):
-        utils.juju_log('INFO', 'ceph: Creating RBD image (%s).' % rbd_img)
-        create_rbd_image(service, pool, rbd_img, sizemb)
-
-    if not image_mapped(rbd_img):
-        utils.juju_log('INFO', 'ceph: Mapping RBD Image as a Block Device.')
-        map_block_storage(service, pool, rbd_img)
-
-    # make file system
-    # TODO: What happens if for whatever reason this is run again and
-    # the data is already in the rbd device and/or is mounted??
-    # When it is mounted already, it will fail to make the fs
-    # XXX: This is really sketchy!  Need to at least add an fstab entry
-    #      otherwise this hook will blow away existing data if its executed
-    #      after a reboot.
-    if not filesystem_mounted(mount_point):
-        make_filesystem(blk_device, fstype)
-
-        for svc in system_services:
-            if utils.running(svc):
-                utils.juju_log('INFO',
-                               'Stopping services %s prior to migrating '
-                               'data' % svc)
-                utils.stop(svc)
-
-        place_data_on_ceph(service, blk_device, mount_point, fstype)
-
-        for svc in system_services:
-            utils.start(svc)
 
=== removed file 'hooks/charmhelpers/lib/cluster_utils.py'
--- hooks/charmhelpers/lib/cluster_utils.py	2014-07-09 12:37:36 +0000
+++ hooks/charmhelpers/lib/cluster_utils.py	1970-01-01 00:00:00 +0000
@@ -1,128 +0,0 @@
-#
-# Copyright 2012 Canonical Ltd.
-#
-# This file is sourced from lp:openstack-charm-helpers
-#
-# Authors:
-#  James Page <[e-mail]>
-#  Adam Gandelman <[e-mail]>
-#
-
-from utils import (
-    juju_log,
-    relation_ids,
-    relation_list,
-    relation_get,
-    get_unit_hostname,
-    config_get)
-import subprocess
-import os
-
-
-def is_clustered():
-    for r_id in (relation_ids('ha') or []):
-        for unit in (relation_list(r_id) or []):
-            clustered = relation_get('clustered',
-                                     rid=r_id,
-                                     unit=unit)
-            if clustered:
-                return True
-    return False
-
-
-def is_leader(resource):
-    cmd = [
-        "crm", "resource",
-        "show", resource]
-    try:
-        status = subprocess.check_output(cmd)
-    except subprocess.CalledProcessError:
-        return False
-    else:
-        if get_unit_hostname() in status:
-            return True
-        else:
-            return False
-
-
-def peer_units():
-    peers = []
-    for r_id in (relation_ids('cluster') or []):
-        for unit in (relation_list(r_id) or []):
-            peers.append(unit)
-    return peers
-
-
-def oldest_peer(peers):
-    local_unit_no = os.getenv('JUJU_UNIT_NAME').split('/')[1]
-    for peer in peers:
-        remote_unit_no = peer.split('/')[1]
-        if remote_unit_no < local_unit_no:
-            return False
-    return True
-
-
-def eligible_leader(resource):
-    if is_clustered():
-        if not is_leader(resource):
-            juju_log('INFO', 'Deferring action to CRM leader.')
-            return False
-    else:
-        peers = peer_units()
-        if peers and not oldest_peer(peers):
-            juju_log('INFO', 'Deferring action to oldest service unit.')
-            return False
-    return True
-
-
-def https():
-    '''
-    Determines whether enough data has been provided in configuration
-    or relation data to configure HTTPS
-    .
-    returns: boolean
-    '''
-    if config_get('use-https') == "yes":
-        return True
-    if config_get('ssl_cert') and config_get('ssl_key'):
-        return True
-    for r_id in relation_ids('identity-service'):
-        for unit in relation_list(r_id):
-            if (relation_get('https_keystone', rid=r_id, unit=unit) and
-                    relation_get('ssl_cert', rid=r_id, unit=unit) and
-                    relation_get('ssl_key', rid=r_id, unit=unit) and
-                    relation_get('ca_cert', rid=r_id, unit=unit)):
-                return True
-    return False
-
-
-def determine_api_port(public_port):
-    '''
-    Determine correct API server listening port based on
-    existence of HTTPS reverse proxy and/or haproxy.
-
-    public_port: int: standard public port for given service
-
-    returns: int: the correct listening port for the API service
-    '''
-    i = 0
-    if len(peer_units()) > 0 or is_clustered():
-        i += 1
-    if https():
-        i += 1
-    return public_port - (i * 10)
-
-
-def determine_haproxy_port(public_port):
-    '''
-    Description: Determine correct proxy listening port based on public IP +
-    existence of HTTPS reverse proxy.
-
-    public_port: int: standard public port for given service
-
-    returns: int: the correct listening port for the HAProxy service
-    '''
-    i = 0
-    if https():
-        i += 1
-    return public_port - (i * 10)
 
=== removed file 'hooks/charmhelpers/lib/utils.py'
--- hooks/charmhelpers/lib/utils.py	2014-07-09 12:37:36 +0000
+++ hooks/charmhelpers/lib/utils.py	1970-01-01 00:00:00 +0000
@@ -1,221 +0,0 @@
-#
-# Copyright 2012 Canonical Ltd.
-#
-# This file is sourced from lp:openstack-charm-helpers
-#
-# Authors:
-#  James Page <[e-mail]>
-#  Paul Collins <[e-mail]>
-#  Adam Gandelman <[e-mail]>
-#
-
-import json
-import os
-import subprocess
-import socket
-import sys
-
-
-def do_hooks(hooks):
-    hook = os.path.basename(sys.argv[0])
-
-    try:
-        hook_func = hooks[hook]
-    except KeyError:
-        juju_log('INFO',
-                 "This charm doesn't know how to handle '{}'.".format(hook))
-    else:
-        hook_func()
-
-
-def install(*pkgs):
-    cmd = [
-        'apt-get',
-        '-y',
-        'install']
-    for pkg in pkgs:
-        cmd.append(pkg)
-    subprocess.check_call(cmd)
-
-TEMPLATES_DIR = 'templates'
-
-try:
-    import jinja2
-except ImportError:
-    install('python-jinja2')
-    import jinja2
-
-try:
-    import dns.resolver
-except ImportError:
-    install('python-dnspython')
-    import dns.resolver
-
-
-def render_template(template_name, context, template_dir=TEMPLATES_DIR):
-    templates = jinja2.Environment(loader=jinja2.FileSystemLoader(
-                                   template_dir))
-    template = templates.get_template(template_name)
-    return template.render(context)
-
-# Protocols
-TCP = 'TCP'
-UDP = 'UDP'
-
-
-def expose(port, protocol='TCP'):
-    cmd = [
-        'open-port',
-        '{}/{}'.format(port, protocol)]
-    subprocess.check_call(cmd)
-
-
-def juju_log(severity, message):
-    cmd = [
-        'juju-log',
-        '--log-level', severity,
-        message]
-    subprocess.check_call(cmd)
-
-
-def relation_ids(relation):
-    cmd = [
-        'relation-ids',
-        relation]
-    result = str(subprocess.check_output(cmd)).split()
-    if result == "":
-        return None
-    else:
-        return result
-
-
-def relation_list(rid):
-    cmd = [
-        'relation-list',
-        '-r', rid]
-    result = str(subprocess.check_output(cmd)).split()
-    if result == "":
-        return None
-    else:
-        return result
-
-
-def relation_get(attribute, unit=None, rid=None):
-    cmd = [
-        'relation-get']
-    if rid:
-        cmd.append('-r')
-        cmd.append(rid)
-    cmd.append(attribute)
-    if unit:
-        cmd.append(unit)
-    value = subprocess.check_output(cmd).strip()  # IGNORE:E1103
-    if value == "":
-        return None
-    else:
-        return value
-
-
-def relation_set(**kwargs):
-    cmd = [
-        'relation-set']
-    args = []
-    for k, v in kwargs.items():
-        if k == 'rid':
-            if v:
-                cmd.append('-r')
-                cmd.append(v)
-        else:
-            args.append('{}={}'.format(k, v))
-    cmd += args
-    subprocess.check_call(cmd)
-
-
-def unit_get(attribute):
-    cmd = [
-        'unit-get',
-        attribute]
-    value = subprocess.check_output(cmd).strip()  # IGNORE:E1103
-    if value == "":
-        return None
-    else:
-        return value
-
-
-def config_get(attribute):
-    cmd = [
-        'config-get',
-        '--format',
-        'json']
-    out = subprocess.check_output(cmd).strip()  # IGNORE:E1103
-    cfg = json.loads(out)
-
-    try:
-        return cfg[attribute]
-    except KeyError:
-        return None
-
-
-def get_unit_hostname():
-    return socket.gethostname()
-
-
-def get_host_ip(hostname=unit_get('private-address')):
-    try:
-        # Test to see if already an IPv4 address
-        socket.inet_aton(hostname)
-        return hostname
-    except socket.error:
-        answers = dns.resolver.query(hostname, 'A')
-        if answers:
-            return answers[0].address
-    return None
-
-
-def _svc_control(service, action):
-    subprocess.check_call(['service', service, action])
-
-
-def restart(*services):
-    for service in services:
-        _svc_control(service, 'restart')
-
-
-def stop(*services):
-    for service in services:
-        _svc_control(service, 'stop')
-
-
-def start(*services):
-    for service in services:
-        _svc_control(service, 'start')
-
-
-def reload(*services):
-    for service in services:
-        try:
-            _svc_control(service, 'reload')
-        except subprocess.CalledProcessError:
-            # Reload failed - either service does not support reload
-            # or it was not running - restart will fixup most things
-            _svc_control(service, 'restart')
-
-
-def running(service):
-    try:
-        output = subprocess.check_output(['service', service, 'status'])
-    except subprocess.CalledProcessError:
-        return False
-    else:
-        if ("start/running" in output or "is running" in output):
-            return True
-        else:
-            return False
-
-
-def is_relation_made(relation, key='private-address'):
-    for r_id in (relation_ids(relation) or []):
-        for unit in (relation_list(r_id) or []):
-            if relation_get(key, rid=r_id, unit=unit):
-                return True
-    return False
 
=== removed file 'hooks/charmhelpers/setup.py'
--- hooks/charmhelpers/setup.py	2014-07-09 12:37:36 +0000
+++ hooks/charmhelpers/setup.py	1970-01-01 00:00:00 +0000
@@ -1,12 +0,0 @@
-#!/usr/bin/env python
-
-from distutils.core import setup
-
-setup(name='charmhelpers',
-     version='1.0',
-     description='this is dumb',
-     author='nobody',
-     author_email='dummy@amulet',
-     url='http://google.com',
-     packages=[],
-)
 
=== modified file 'hooks/hdp-hadoop-common.py'
--- hooks/hdp-hadoop-common.py	2014-11-21 19:35:29 +0000
+++ hooks/hdp-hadoop-common.py	2015-02-13 00:38:49 +0000
@@ -1,19 +1,28 @@
 #!/usr/bin/env python
 import os
+import re
 import subprocess
+import pwd
 import sys
 import tarfile
 import shlex
 import shutil
 import inspect
 import time
+from functools import partial
+import socket
+
+import bootstrap
+bootstrap.install_charmhelpers()
 
 from hdputils import install_base_pkg, updateHDPDirectoryScript, config_all_nodes, \
-                    setHadoopEnvVar, home, hdpScript, configureJAVA, config_all_nodes
-from bdutils import setDirPermission, fileSetKV, append_bashrc, is_jvm_service_active, setHadoopConfigXML, chownRecursive
+                    setHadoopEnvVar, home, hdpScript, configureJAVA, config_all_nodes, \
+                    javaPath
+from bdutils import setDirPermission, fileSetKV, append_bashrc, is_jvm_service_active, setHadoopConfigXML, \
+                    chownRecursive, HDFS_command, fconfigured
 
-from charmhelpers.lib.utils import config_get, get_unit_hostname
-from shutil import rmtree, copyfile
+from shutil import rmtree, copyfile, copy
+from charmhelpers.core import hookenv
 from charmhelpers.core.hookenv import log, Hooks, relation_get, relation_set, unit_get, open_port, local_unit, related_units
 from charmhelpers.core.host import service_start, service_stop, add_user_to_group
 from time import sleep
@@ -68,27 +77,71 @@
 # in config.yaml
 def updateDirectoriesScript():
     log("==> {}".format(inspect.stack()[0][3]),"INFO")
-    fileSetKV(directoriesScript, "DFS_NAME_DIR=", "\""+config_get('dfs_name_dir')+"\";")
-    fileSetKV(directoriesScript, "DFS_DATA_DIR=","\""+ config_get('dfs_data_dir')+"\";")
-    fileSetKV(directoriesScript, "FS_CHECKPOINT_DIR=", "\""+config_get('fs_checkpoint_dir')+"\";")
-    fileSetKV(directoriesScript, "YARN_LOCAL_DIR=", "\""+config_get('yarn_local_dir')+"\";")
-    fileSetKV(directoriesScript, "YARN_LOCAL_LOG_DIR=", "\""+config_get('yarn_local_log_dir')+"\";")
-    fileSetKV(directoriesScript, "ZOOKEEPER_DATA_DIR=", "\""+config_get('zookeeper_data_dir')+"\";")
+    fileSetKV(directoriesScript, "DFS_NAME_DIR=", "\""+hookenv.config('dfs_name_dir')+"\";")
+    fileSetKV(directoriesScript, "DFS_DATA_DIR=","\""+ hookenv.config('dfs_data_dir')+"\";")
+    fileSetKV(directoriesScript, "FS_CHECKPOINT_DIR=", "\""+hookenv.config('fs_checkpoint_dir')+"\";")
+    fileSetKV(directoriesScript, "YARN_LOCAL_DIR=", "\""+hookenv.config('yarn_local_dir')+"\";")
+    fileSetKV(directoriesScript, "YARN_LOCAL_LOG_DIR=", "\""+hookenv.config('yarn_local_log_dir')+"\";")
+    fileSetKV(directoriesScript, "ZOOKEEPER_DATA_DIR=", "\""+hookenv.config('zookeeper_data_dir')+"\";")
 
 
 def createHDPHadoopConf():
     log("==> {}".format(inspect.stack()[0][3]),"INFO")
 
+    edits = {
+        'hadoop-env.sh': {
+            r'TODO-JDK-PATH': javaPath,
+        },
+        'yarn-env.sh': {
+            r'TODO-JDK-PATH': javaPath,
+        },
+    }
+
     HADOOP_CONF_DIR = os.environ["HADOOP_CONF_DIR"]
     HDPConfPath = os.path.join(os.path.sep,home, hdpScript, "configuration_files", "core_hadoop")
     source = os.listdir(HDPConfPath)
     for files in source:
         srcFile = os.path.join(os.path.sep, HDPConfPath, files)
         desFile = os.path.join(os.path.sep, HADOOP_CONF_DIR, files)
-        shutil.copyfile(srcFile, desFile)
+        edit_file(srcFile, desFile, edits.get(files))
     subprocess.call(createDadoopConfDir)
 
 
+def edit_file(src, dst=None, edits=None):
+    """
+    Copy file from src to dst, applying one or more per-line edits.
+
+    :param str src: Path of source file to copy from
+    :param str dst: Path of destination file.  If None or the same as `src`,
+                    the edits will be made in-place.
+    :param dict edits: A mapping of regular expression patterns to regular
+                       expression replacement patterns (using backreferences).
+                       Any matching replacements will be made for each line.
+                       If the order of replacements is important, an instance
+                       of the `OrderedDict` class from the `collections`
+                       package should be used.
+    """
+    if not edits:
+        if not dst or dst == src:
+            return  # in-place edit with no changes is a noop
+        shutil.copyfile(src, dst)  # no edits is just a copy
+        return
+    cleanup = False
+    if not dst or dst == src:
+        dst = src
+        src = '{}.bak'.format(src)
+        os.rename(dst, src)
+        cleanup = True
+    with open(src, 'r') as infile:
+        with open(dst, 'w') as outfile:
+            for line in infile.readlines():
+                for pat, repl in edits.iteritems():
+                    line = re.sub(pat, repl, line)
+                outfile.write(line)
+    if cleanup:
+        os.remove(src)
+
+
 def uninstall_base_pkg():
     log("==> {}".format(inspect.stack()[0][3]),"INFO")
     packages = ['ntp',
@@ -104,7 +157,6 @@
                 'liblzo2-2',
                 'liblzo2-dev',
                 'libhdfs0',
-                'libhdfs0-dev',
                 'hadoop-lzo']
     apt_purge(packages)
     shutil.rmtree(hdpScriptPath)
@@ -130,26 +182,41 @@
     setDirPermission(os.environ['MAPRED_PID_DIR'], os.environ['MAPRED_USER'], group, 0755)
     setDirPermission(os.environ['YARN_LOCAL_DIR'], os.environ['YARN_USER'], group, 0755)
     setDirPermission(os.environ['YARN_LOCAL_LOG_DIR'], os.environ['YARN_USER'], group, 0755)
-    hdfsConfPath = os.path.join(os.path.sep, os.environ['HADOOP_CONF_DIR'],'hdfs-site.xml')
-    setHadoopConfigXML(hdfsConfPath, "dfs.namenode.name.dir", config_get('dfs_name_dir'))
-    setHadoopConfigXML(hdfsConfPath, "dfs.datanode.data.dir", config_get('dfs_data_dir'))
-    setHadoopConfigXML(hdfsConfPath, "dfs.namenode.checkpoint.dir", config_get('fs_checkpoint_dir'))
+    hdfsConfPath = os.path.join(os.path.sep, os.environ['HADOOP_CONF_DIR'], 'hdfs-site.xml')
+    setHadoopConfigXML(hdfsConfPath, "dfs.namenode.name.dir", hookenv.config('dfs_name_dir'))
+    setHadoopConfigXML(hdfsConfPath, "dfs.datanode.data.dir", hookenv.config('dfs_data_dir'))
+    setHadoopConfigXML(hdfsConfPath, "dfs.namenode.checkpoint.dir", hookenv.config('fs_checkpoint_dir'))
+    yarnConfPath = os.path.join(os.path.sep, os.environ['HADOOP_CONF_DIR'], 'yarn-site.xml')
+    setHadoopConfigXML(yarnConfPath, "yarn.scheduler.maximum-allocation-mb", "2048")
+    setHadoopConfigXML(yarnConfPath, "yarn.scheduler.minimum-allocation-mb", "682")
+    setHadoopConfigXML(yarnConfPath, "yarn.nodemanager.resource.cpu-vcores", "2")
+    setHadoopConfigXML(yarnConfPath, "yarn.nodemanager.resource.memory-mb", "2048")
+    setHadoopConfigXML(yarnConfPath, "yarn.nodemanager.resource.percentage-physical-cpu-limit", "100")
+    mapredConfPath = os.path.join(os.path.sep, os.environ['HADOOP_CONF_DIR'], 'mapred-site.xml')
+#    setHadoopConfigXML(mapredConfPath, "mapreduce.application.framework.path", None)
+    setHadoopConfigXML(mapredConfPath, "mapreduce.map.memory.mb", "682")
+    setHadoopConfigXML(mapredConfPath, "mapreduce.reduce.memory.mb", "682")
+    setHadoopConfigXML(mapredConfPath, "mapreduce.map.java.opts", "-Xmx546m")
+    setHadoopConfigXML(mapredConfPath, "mapreduce.reduce.java.opts", "-Xmx546m")
+    setHadoopConfigXML(mapredConfPath, "mapreduce.task.io.sort.factor", "100")
+    setHadoopConfigXML(mapredConfPath, "mapreduce.task.io.sort.mb", "273")
+    setHadoopConfigXML(mapredConfPath, "yarn.app.mapreduce.am.resource.mb", "682")
 
 # candidate for BD charm helper
 def format_namenode(hdfsUser):
     log("==> hdfs format for user={}".format(hdfsUser),"INFO")
-    cmd = shlex.split("su {} -c '/usr/lib/hadoop/bin/hadoop namenode -format'".format(hdfsUser))
+    cmd = shlex.split("su {} -c '/usr/hdp/current/hadoop-client/bin/hadoop namenode -format'".format(hdfsUser))
     subprocess.call(cmd)
 
 def callHDFS_fs(command):
-    cmd = shlex.split("su hdfs -c '/usr/lib/hadoop/bin/hadoop fs {}'".format(command))
+    cmd = shlex.split("su hdfs -c '/usr/hdp/current/hadoop-client/bin/hadoop fs {}'".format(command))
     subprocess.call(cmd)
 
 
 ''' Start Jobhistory Server '''
 def start_jobhistory():
     log("==> Start Job History Server")
-    path = os.path.join(os.path.sep, 'usr', 'lib', 'hadoop-yarn', 'bin', 'container-executor')
+    path = '/usr/hdp/current/hadoop-yarn-client/bin/container-executor'
     chownRecursive(path, 'root', 'hadoop')
     os.chmod(path, 650)
     callHDFS_fs("-mkdir -p /mr-history/tmp")
@@ -161,16 +228,16 @@
     callHDFS_fs("-chmod -R 1777 /app-logs ")
     callHDFS_fs("-chown yarn /app-logs ")
     hadoopConfDir = os.environ["HADOOP_CONF_DIR"]
-    os.environ["HADOOP_LIBEXEC_DIR"]="/usr/lib/hadoop/libexec"
-    cmd = shlex.split("su {} -c '/usr/lib/hadoop-mapreduce/sbin/mr-jobhistory-daemon.sh --config {} start historyserver'".\
+    os.environ["HADOOP_LIBEXEC_DIR"] = "/usr/hdp/current/hadoop-client/libexec"
+    cmd = shlex.split("su {} -c '/usr/hdp/current/hadoop-mapreduce-client/sbin/mr-jobhistory-daemon.sh --config {} start historyserver'".\
                       format(os.environ['MAPRED_USER'], hadoopConfDir))
     subprocess.call(cmd)
 
 '''Stop Job History Server'''
 def stop_jobhistory():
     hadoopConfDir = os.environ["HADOOP_CONF_DIR"]
-    os.environ["HADOOP_LIBEXEC_DIR"]="/usr/lib/hadoop/libexec"
-    cmd = shlex.split("su {} -c '/usr/lib/hadoop-mapreduce/sbin/mr-jobhistory-daemon.sh --config {} stop historyserver'".\
+    os.environ["HADOOP_LIBEXEC_DIR"] = "/usr/hdp/current/hadoop-client/libexec"
+    cmd = shlex.split("su {} -c '/usr/hdp/current/hadoop-mapreduce-client/sbin/mr-jobhistory-daemon.sh --config {} stop historyserver'".\
                       format(os.environ['MAPRED_USER'], hadoopConfDir))
     subprocess.call(cmd)
 
@@ -182,47 +249,49 @@
        stop_jobhistory()
        start_jobhistory()
 
-''' Start Namenode Service
-       hdfsUser: System user for Hadoop Distributed Filesystem
-'''
+
+def hdfs_service(service, action, username):
+    daemon = '/usr/hdp/current/hadoop-hdfs-{}/../hadoop/sbin/hadoop-daemon.sh'
+    log("==> {} {} for user={}".format(action, service, username), "INFO")
+    subprocess.check_call(shlex.split(
+        "su {user} -c '{daemon} --config {conf_dir} {action} {service}'".format(
+            user=username,
+            daemon=daemon.format(service),
+            service=service,
+            action=action,
+            conf_dir=os.environ["HADOOP_CONF_DIR"])))
+
+
 def start_namenode(hdfsUser):
-    log("==> start namenode for user={}".format(hdfsUser), "INFO")
-    hadoopConfDir = os.environ["HADOOP_CONF_DIR"]
-    cmd = shlex.split("su {} -c '/usr/lib/hadoop/sbin/hadoop-daemon.sh --config {} start namenode'".\
-                      format(hdfsUser, hadoopConfDir))
-    subprocess.check_call(cmd)
-
-'''Stop Name Node server
-    hdfsUser: System user for Hadoop Distributed Filesystem
-'''
+    '''
+    Start Namenode Service
+        hdfsUser: System user for Hadoop Distributed Filesystem
+    '''
+    hdfs_service('namenode', 'start', hdfsUser)
+
+
 def stop_namenode(hdfsUser):
-    log("==> stop namenode for user={}".format(hdfsUser), "INFO")
-    hadoopConfDir = os.environ["HADOOP_CONF_DIR"]
-    cmd = shlex.split("su {} -c '/usr/lib/hadoop/sbin/hadoop-daemon.sh --config {} stop namenode'".\
-                      format(hdfsUser, hadoopConfDir))
-    subprocess.call(cmd)
-
-'''
- Start Data Node
-    hdfsUser: System user for Hadoop Distributed Filesystem
-'''
+    '''
+    Stop Name Node server
+        hdfsUser: System user for Hadoop Distributed Filesystem
+    '''
+    hdfs_service('namenode', 'stop', hdfsUser)
+
+
 def start_datanode(hdfsUser):
-    log("==> start datanode for user={}".format(hdfsUser), "INFO")
-    hadoopConfDir = os.environ["HADOOP_CONF_DIR"]
-    cmd = shlex.split("su {} -c '/usr/lib/hadoop/sbin/hadoop-daemon.sh --config {} start datanode'".\
-                      format(hdfsUser, hadoopConfDir))
-    subprocess.check_call(cmd)
-
-'''
- Stop Data Node
-    hdfsUser: System user for Hadoop Distributed Filesystem
-'''
+    '''
+    Start Data Node
+        hdfsUser: System user for Hadoop Distributed Filesystem
+    '''
+    hdfs_service('datanode', 'start', hdfsUser)
+
+
 def stop_datanode(hdfsUser):
-    log("==> stop datanode for user={}".format(hdfsUser), "INFO")
-    hadoopConfDir = os.environ["HADOOP_CONF_DIR"]
-    cmd = shlex.split("su {} -c '/usr/lib/hadoop/sbin/hadoop-daemon.sh --config {} stop datanode'".\
-                      format(hdfsUser, hadoopConfDir))
-    subprocess.call(cmd)
+    '''
+    Stop Data Node
+        hdfsUser: System user for Hadoop Distributed Filesystem
+    '''
+    hdfs_service('datanode', 'stop', hdfsUser)
 # candidate for BD charm helper
 
 '''
@@ -238,52 +307,49 @@
     setHadoopConfigXML(yarnConfPath, "yarn.resourcemanager.resource-tracker.address", RMhostname+":8025")
     setHadoopConfigXML(yarnConfPath, "yarn.resourcemanager.scheduler.address", RMhostname+":8030")
     setHadoopConfigXML(yarnConfPath, "yarn.resourcemanager.address", RMhostname+":8050")
+    setHadoopConfigXML(yarnConfPath, "yarn.resourcemanager.hostname", RMhostname)
     setHadoopConfigXML(yarnConfPath, "yarn.resourcemanager.admin.address", RMhostname+":8141")
-    setHadoopConfigXML(yarnConfPath, "yarn.nodemanager.local-dirs", config_get('yarn_local_dir'))
-    setHadoopConfigXML(yarnConfPath, "yarn.nodemanager.log-dirs", config_get('yarn_local_log_dir'))
+    setHadoopConfigXML(yarnConfPath, "yarn.nodemanager.local-dirs", hookenv.config('yarn_local_dir'))
+    setHadoopConfigXML(yarnConfPath, "yarn.nodemanager.log-dirs", hookenv.config('yarn_local_log_dir'))
     setHadoopConfigXML(yarnConfPath, "yarn.log.server.url","http://"+RMhostname+":19888")
     setHadoopConfigXML(yarnConfPath, "yarn.resourcemanager.webapp.address", RMhostname+":8088")
     #jobhistory server
     setHadoopConfigXML(mapConfDir, "mapreduce.jobhistory.webapp.address", RMhostname+":19888")
     setHadoopConfigXML(mapConfDir, "mapreduce.jobhistory.address",  RMhostname+":10020")
 
-''' Start Resource Manager server '''
+
+def yarn_service(service, action, username):
+    daemon = '/usr/hdp/current/hadoop-yarn-client/sbin/yarn-daemon.sh'
+    log("==> {} {} for user={}".format(action, service, username), "INFO")
+    os.environ["HADOOP_LIBEXEC_DIR"] = "/usr/hdp/current/hadoop-client/libexec"
+    subprocess.check_call(shlex.split(
+        "su {user} -c '{daemon} --config {conf_dir} {action} {service}'".format(
+            user=username,
+            daemon=daemon.format(service),
+            service=service,
+            action=action,
+            conf_dir=os.environ["HADOOP_CONF_DIR"])))
+
+
 def start_resourcemanager(yarnUser):
-    log("==> start resourcemanager", "INFO")
-    hadoopConfDir = os.environ["HADOOP_CONF_DIR"]
-    os.environ["HADOOP_LIBEXEC_DIR"]="/usr/lib/hadoop/libexec"
-    cmd = shlex.split("su {} -c '/usr/lib/hadoop-yarn/sbin/yarn-daemon.sh --config {} start resourcemanager'".\
-                      format(yarnUser, hadoopConfDir))
-    subprocess.call(cmd)
-
-''' Stop Resource Manager server '''
+    ''' Start Resource Manager server '''
+    yarn_service('resourcemanager', 'start', yarnUser)
+
+
 def stop_resourcemanager(yarnUser):
-    log("==> stop resourcemanager", "INFO")
-    hadoopConfDir = os.environ["HADOOP_CONF_DIR"]
-    os.environ["HADOOP_LIBEXEC_DIR"]="/usr/lib/hadoop/libexec"
-    cmd = shlex.split("su {} -c '/usr/lib/hadoop-yarn/sbin/yarn-daemon.sh --config {} stop resourcemanager'".\
-                      format(yarnUser, hadoopConfDir))
-    subprocess.check_call(cmd)
-
-''' Start Node Manager daemon on each compute node '''
+    ''' Stop Resource Manager server '''
+    yarn_service('resourcemanager', 'stop', yarnUser)
+
+
 def start_nodemanager(yarnUser):
-    log("==> start nodemanager", "INFO")
-    hadoopConfDir = os.environ["HADOOP_CONF_DIR"]
-    os.environ["HADOOP_LIBEXEC_DIR"]="/usr/lib/hadoop/libexec"
-    cmd = shlex.split("su {} -c '/usr/lib/hadoop-yarn/sbin/yarn-daemon.sh --config {} start nodemanager'".\
-                      format(yarnUser, hadoopConfDir))
-    subprocess.check_call(cmd)
-
-
-''' Stop Node Manager daemon on each compute node '''
+    ''' Start Node Manager daemon on each compute node '''
+    yarn_service('nodemanager', 'start', yarnUser)
+
 
 def stop_nodemanager(yarnUser):
-    log("==> stop nodemanager", "INFO")
-    hadoopConfDir = os.environ["HADOOP_CONF_DIR"]
-    os.environ["HADOOP_LIBEXEC_DIR"]="/usr/lib/hadoop/libexec"
-    cmd = shlex.split("su {} -c '/usr/lib/hadoop-yarn/sbin/yarn-daemon.sh --config {} stop nodemanager'".\
-                      format(yarnUser, hadoopConfDir))
-    subprocess.call(cmd)
+    ''' Stop Node Manager daemon on each compute node '''
+    yarn_service('nodemanager', 'stop', yarnUser)
+
 
 '''Stop all running hadoop services
       NOTE: Order is important - DO NOT CHANGE
@@ -336,6 +402,8 @@
         break
 
 def configureHDFS(hostname):
+    cmd = "/usr/bin/hdp-select set all 2.2.0.0-2041"
+    subprocess.check_call(shlex.split(cmd))
     hdfsConfPath = os.path.join(os.path.sep, os.environ['HADOOP_CONF_DIR'],'hdfs-site.xml')
     coreConfPath = os.path.join(os.path.sep, os.environ['HADOOP_CONF_DIR'],'core-site.xml')
     setHadoopConfigXML(coreConfPath, "fs.defaultFS", "hdfs://"+hostname+":8020")
@@ -361,11 +429,10 @@
                 'liblzo2-2',
                 'liblzo2-dev',
                 'libhdfs0',
-                'libhdfs0-dev',
                 'hadoop-lzo']
     install_base_pkg(packages)
     config_hadoop_nodes()
-    fileSetKV(hosts_path, unit_get('private-address')+' ', get_unit_hostname())
+    fileSetKV(hosts_path, unit_get('private-address')+' ', socket.gethostname())
 
 
 @hooks.hook('resourcemanager-relation-joined')
@@ -373,17 +440,23 @@
     log ("==> resourcemanager-relation-joined","INFO")
     if is_jvm_service_active("ResourceManager"):
         relation_set(resourceManagerReady=True)
-        relation_set(resourceManager_hostname=get_unit_hostname())
+        relation_set(resourceManager_hostname=socket.gethostname())
         return
+    # waiting for namenode, however this will not work on a distributed hdfs system. 
     if not is_jvm_service_active("NameNode"):
         sys.exit(0)
+    shutil.copy(os.path.join(os.path.sep, os.environ['CHARM_DIR'],\
+                             'files', 'scripts', "terasort.sh"), home)
     setHadoopEnvVar()
     relation_set(resourceManager_ip=unit_get('private-address'))
-    relation_set(resourceManager_hostname=get_unit_hostname())
+    relation_set(resourceManager_hostname=socket.gethostname())
     configureYarn(unit_get('private-address'))
     start_resourcemanager(os.environ["YARN_USER"])
     start_jobhistory()
     open_port(8025)
+    open_port(8050)
+    open_port(8020)
+    open_port(8042)
     open_port(8030)
     open_port(8050)
     open_port(8141)
@@ -414,7 +487,9 @@
     open_port(19888)
     open_port(8088)
     open_port(10020)
-    relation_set(nodemanager_hostname=get_unit_hostname())
+    open_port(8042)
+
+    relation_set(nodemanager_hostname=socket.gethostname())
 
 
 @hooks.hook('resourcemanager-relation-broken')
@@ -437,25 +512,32 @@
 
     if is_jvm_service_active("NameNode"):
         relation_set(nameNodeReady=True)
-        relation_set(namenode_hostname=get_unit_hostname())
+        relation_set(namenode_hostname=socket.gethostname())
         return
     setHadoopEnvVar()
     setDirPermission(os.environ['DFS_NAME_DIR'], os.environ['HDFS_USER'], os.environ['HADOOP_GROUP'], 0755)
-    relation_set(namenode_hostname=get_unit_hostname())
+    relation_set(namenode_hostname=socket.gethostname())
     configureHDFS(unit_get('private-address'))
     format_namenode(os.environ["HDFS_USER"])
     start_namenode(os.environ["HDFS_USER"])
+    # Namenode might not be ready yet
+    sleep(15)
+    # Create hdfs (superuser) and ubunut (default user)
+    HDFS_command("dfs -mkdir -p /user/hdfs")
+    HDFS_command("dfs -mkdir -p /user/ubuntu")
+    HDFS_command("dfs -chown ubuntu /user/ubuntu")
+    HDFS_command("dfs -chmod -R 755 /user/ubuntu")
     start_jobhistory()
     open_port(8020)
     open_port(8010)
     open_port(50070)
     open_port(50075)
     open_port(8480)
-    open_port(50470)
+    open_port(50070)
+    open_port(45454)
     if not is_jvm_service_active("NameNode"):
         log("error ==> NameNode failed to start")
         sys.exit(1)
-    sleep(5)
     relation_set(nameNodeReady=True)
 
 
@@ -480,7 +562,7 @@
     open_port(8480)
     open_port(50010)
     open_port(50075)
-    relation_set(datanode_hostname = get_unit_hostname())
+    relation_set(datanode_hostname = socket.gethostname())
 
 
 @hooks.hook('config-changed')
@@ -499,6 +581,15 @@
         sys.exit(0)
     log("Configuring namenode - changed phase", "INFO")
     fileSetKV(hosts_path, relation_get('private-address')+' ', datanode_host)
+    
+    # Upload the MapReduce tarball to HDFS.
+    if not fconfigured("mapred_init"):
+        HDFS_command("dfs -mkdir -p /hdp/apps/2.2.0.0-2041/mapreduce/")
+        HDFS_command("dfs -put /usr/hdp/current/hadoop-client/mapreduce.tar.gz /hdp/apps/2.2.0.0-2041/mapreduce/")
+        HDFS_command("dfs -chown -R hdfs:hadoop /hdp")
+        HDFS_command("dfs -chmod -R 555 /hdp/apps/2.2.0.0-2041/mapreduce")
+        HDFS_command("dfs -chmod -R 444 /hdp/apps/2.2.0.0-2041/mapreduce/mapreduce.tar.gz")
+    
 
 @hooks.hook('start')
 def start():
@@ -525,14 +616,14 @@
 
 @hooks.hook('compute-nodes-relation-joined')
 def compute_nodes_relation_joined():
-    log("==> compute_nodes_relation_joined {}".format(get_unit_hostname()),"INFO")
-    relation_set(hostname=get_unit_hostname())
+    log("==> compute_nodes_relation_joined {}".format(socket.gethostname()),"INFO")
+    relation_set(hostname=socket.gethostname())
 
 
 @hooks.hook('hadoop-nodes-relation-joined')
 def hadoop_nodes_relation_joined():
-    log("==> hadoop_nodes_relation_joined {}".format(get_unit_hostname()),"INFO")
-    relation_set(hostname=get_unit_hostname())
+    log("==> hadoop_nodes_relation_joined {}".format(socket.gethostname()),"INFO")
+    relation_set(hostname=socket.gethostname())
 
 
 
@@ -555,7 +646,5 @@
 resourceManagerReady = False
 
 
-
-
 if __name__ == "__main__":
     hooks.execute(sys.argv)
 
=== modified file 'hooks/hdputils.py'
--- hooks/hdputils.py	2014-11-21 18:37:23 +0000
+++ hooks/hdputils.py	2015-02-13 00:38:49 +0000
@@ -17,24 +17,27 @@
 
 def install_base_pkg(packages):
     log ("==> install_base_pkg", "INFO")
-    wgetPkg("http://public-repo-1.hortonworks.com/HDP/ubuntu12/2.1.3.0/hdp.list -O /etc/apt/sources.list.d/hdp.list","")
-    cmd =gpg_script 
+    wgetPkg("http://public-repo-1.hortonworks.com/HDP/ubuntu12/2.x/GA/2.2.0.0/hdp.list",
+            "/etc/apt/sources.list.d/hdp.list")
+    cmd = gpg_script
     subprocess.call(cmd)
     apt_update()
     apt_install(packages)
     if not os.path.isdir(os.path.join(os.path.sep,'usr','lib', 'hadoop')):
         log("Error, apt-get install Hadoop failed", "ERROR")
         sys.exit(1)
-    os.chdir(home);
-    wgetPkg("http://public-repo-1.hortonworks.com/HDP/tools/2.1.1.0/hdp_manual_install_rpm_helper_files-2.1.1.385.tar.gz","")
-    if tarfile.is_tarfile(tarfilename):
-        tball = tarfile.open(tarfilename)
+    os.chdir(home)
+    wgetPkg("http://public-repo-1.hortonworks.com/HDP/tools/2.2.0.0/hdp_manual_install_rpm_helper_files-2.2.0.0.2041.tar.gz",
+            tarfilename)
+    if not tarfile.is_tarfile(tarfilename):
+        log.error('Unable to extract Hadoop Tarball')
+        sys.exit(1)
+    with tarfile.open(tarfilename) as tball:
+        extracted_dirname = tball.next().name
         tball.extractall(home)
-    else:
-        log ("Unable to extract Hadoop Tarball ", "ERROR")
     if os.path.isdir(hdpScript):
         shutil.rmtree(hdpScript)
-    os.rename(tarfilenamePre, hdpScript)
+    os.rename(extracted_dirname, hdpScript)
     log("<== install_base_pkg", "INFO")
 
 def uninstall_base_pkg(packages):
@@ -116,10 +119,9 @@
 ################ Global values #########################
 home  = os.path.join(os.path.sep, "home", "ubuntu")
 javaPath = "/usr/lib/jvm/java-1.7.0-openjdk-amd64"
-tarfilename="hdp_manual_install_rpm_helper_files-2.1.1.385.tar.gz"
-tarfilenamePre="hdp_manual_install_rpm_helper_files-2.1.1.385"
+tarfilename="/tmp/hdp-helpers.tar.gz"
 HDP_PGP_SCRIPT = 'gpg_ubuntu.sh'
-gpg_script = os.path.join(os.path.sep, os.environ['CHARM_DIR'], os.path.sep, os.environ['CHARM_DIR'], 'files', 'scripts',HDP_PGP_SCRIPT)
+gpg_script = os.path.join(os.environ['CHARM_DIR'], 'files', 'scripts', HDP_PGP_SCRIPT)
 hdpScript = "hdp_scripts"
 hdpScriptPath = os.path.join(os.path.sep,home, hdpScript,'scripts')
 usersAndGroupsScript = os.path.join(os.path.sep, hdpScriptPath, "usersAndGroups.sh")
 
=== added file 'resources.yaml'
--- resources.yaml	1970-01-01 00:00:00 +0000
+++ resources.yaml	2015-02-13 00:38:49 +0000
@@ -0,0 +1,11 @@
+resources:
+  pathlib:
+    pypi: path.py>=7.0
+  pyaml:
+    pypi: pyaml
+  six:
+    pypi: six
+  charmhelpers:
+    pypi: http://bazaar.launchpad.net/~bigdata-dev/bigdata-data/trunk/download/cory.johns%40canonical.com-20150212185722-5i0c57g7725yw71r/charmhelpers0.2.2.ta-20150203190143-yrb1xsbp2bffbyp5-1/charmhelpers-0.2.2.tar.gz
+    hash: 4a6daacc352f7381b38e3efffe760f6c43a15452bdaf66e50f234ea425b91873
+    hash_type: sha256
 
=== modified file 'tests/01-hadoop-cluster-deployment-1.py'
--- tests/01-hadoop-cluster-deployment-1.py	2014-11-21 18:37:23 +0000
+++ tests/01-hadoop-cluster-deployment-1.py	2015-02-13 00:38:49 +0000
@@ -1,141 +1,85 @@
 #!/usr/bin/python3
 
+import unittest
 import amulet
 import yaml
 import os
-class TestDeployment(object):
-    def __init__(self):
-        self.d =  amulet.Deployment(series='trusty')
-        bpath = os.path.join(os.path.dirname( __file__), "hadoop_cluster.yaml")
+
+
+class TestDeployment(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.d = amulet.Deployment(series='trusty')
+        bpath = os.path.join(os.path.dirname(__file__), "hadoop_cluster.yaml")
         with open(bpath) as f:
             bun = f.read()
-        self.d.load(yaml.safe_load(bun))
+        cls.d.load(yaml.safe_load(bun))
         try:
-            self.d.setup(timeout=9000)
-            self.d.sentry.wait()
+            cls.d.setup(timeout=9000)
+            cls.d.sentry.wait()
         except amulet.helpers.TimeoutError:
             amulet.raise_status(amulet.SKIP, msg="Environment wasn't stood up in time")
         except:
             raise
-        self.master_unit = self.d.sentry.unit['yarn-hdfs-master/0']
-        self.compute_unit = self.d.sentry.unit['compute-node/0']
+        cls.master_unit = cls.d.sentry.unit['yarn-hdfs-master/0']
+        cls.compute_unit = cls.d.sentry.unit['compute-node/0']
 
-    def run(self):
-        for test in dir(self):
-            if test.startswith('test_'):
-                getattr(self, test)()
-############################################################
-# Validate hadoop services on master node have been started
-############################################################
     def test_hadoop_master_service_status(self):
-        o,c= self.master_unit.run("jps | awk '{print $2}'")
-        if o.find('ResourceManager') == -1:
-            amulet.raise_status(amulet.FAIL, msg="ResourceManager not started")
-        else:
-            amulet.raise_status(amulet.PASS, msg="ResourceManager started")
-        if o.find('JobHistoryServer') == -1:
-            amulet.raise_status(amulet.FAIL, msg="JobHistoryServer not started")
-        else:
-            amulet.raise_status(amulet.PASS, msg="JobHistoryServer started")
-        if o.find('NameNode') == -1:
-            amulet.raise_status(amulet.FAIL, msg="NameNode not started")
-        else:
-            amulet.raise_status(amulet.PASS, msg="NameNode started")
+        """
+        Validate hadoop services on master node have been started
+        """
+        output, retcode = self.master_unit.run("pgrep -a java")
+        assert 'ResourceManager' in output, "ResourceManager not started"
+        assert 'JobHistoryServer' in output, "JobHistoryServer not started"
+        assert 'NameNode' in output, "NameNode not started"
 
-############################################################
-# Validate hadoop services on compute node have been started
-############################################################
     def test_hadoop_compute_service_status(self):
-        o,c= self.compute_unit.run("jps | awk '{print $2}'")
-        if o.find('NodeManager') == -1:
-            amulet.raise_status(amulet.FAIL, msg="NodeManager not started")
-        else:
-            amulet.raise_status(amulet.PASS, msg="NodeManager started")
-        if o.find('DataNode') == -1:
-            amulet.raise_status(amulet.FAIL, msg="DataServer not started")
-        else:
-            amulet.raise_status(amulet.PASS, msg="DataServer started")
-###########################################################################
-# Validate admin few hadoop activities on HDFS cluster.
-###########################################################################
-# 1) This test validates mkdir on hdfs cluster
-###########################################################################
-    def test_hdfs_mkdir(self):
-        o,c= slef.master_unit.run("su hdfs -c 'hdfs dfs -mkdir -p /user/ubuntu'")
-        if c == 0:
-            amulet.raise_status(amulet.PASS, msg=" Successfully created a user directory on hdfs")
-        else:
-            amulet.raise_status(amulet.FAIL, msg=" Created a user directory on hdfs FAILED")
-###########################################################################
-# 2) This test validates change hdfs dir owner on the cluster
-###########################################################################
-    def test_hdfs_dir_owner(self):
-        o,c= self.master_unit.run("su hdfs -c 'hdfs dfs -chown ubuntu:ubuntu /user/ubuntu'")
-        if c == 0:
-            amulet.raise_status(amulet.PASS, msg=" Successfully assigned an owner to directory on hdfs")
-        else:
-            amulet.raise_status(amulet.FAIL, msg=" assigning an owner to hdfs directory FAILED")
-
-###########################################################################
-# 3) This test validates setting hdfs directory access permission on the cluster
-###########################################################################
-    def test_hdfs_dir_permission(self):
-        o,c= self.master_unit.run("su hdfs -c 'hdfs dfs -chmod -R 755 /user/ubuntu'")
-        if c == 0:
-            amulet.raise_status(amulet.PASS, msg=" Successfully set directory permission on hdfs")
-        else:
-            amulet.raise_status(amulet.FAIL, msg=" seting directory permission on hdfs FAILED")
-
-###########################################################################
-# Validate yarn mapreduce operation
-# 1) validate mapreduce execution - writing to hdfs
-# 2) validate successful mapreduce operation after the execution
-###########################################################################
-    def test_yarn_mapreduce_exe1(self):
-        o,c= self.master_unit.run("su ubuntu -c 'hadoop jar /usr/lib/hadoop-mapreduce/hadoop-mapreduce-examples-*.jar teragen  10000 /user/ubuntu/teragenout'")
-        if c == 0:
-            amulet.raise_status(amulet.PASS, msg=" Successfull execution of teragen mapreduce app")
-        else:
-            amulet.raise_status(amulet.FAIL, msg=" teragen FAILED")
-###########################################################################
-# 2) validate successful mapreduce operation after the execution
-###########################################################################
-        o,c= self.master_unit.run("su hdfs -c 'hdfs dfs -ls /user/ubuntu/teragenout/_SUCCESS'")
-        if c == 0:
-            amulet.raise_status(amulet.PASS, msg="mapreduce operation was success")
-        else:
-            amulet.raise_status(amulet.FAIL, msg="mapreduce operation was not a success")
-
-###########################################################################
-# Validate yarn mapreduce operation
-# 1) validate mapreduce execution - reading and writing to hdfs
-# 2) validate successful mapreduce operation after the execution
-###########################################################################
-    def test_yarn_mapreduce_exe2(self):
-        o,c= self.master_unit.run("su ubuntu -c 'hadoop jar /usr/lib/hadoop-mapreduce/hadoop-mapreduce-examples-*.jar terasort /user/ubuntu/teragenout /user/ubuntu/terasortout'")
-        if c == 0:
-            amulet.raise_status(amulet.PASS, msg=" Successfull execution of terasort mapreduce app")
-        else:
-            amulet.raise_status(amulet.FAIL, msg=" terasort FAILED")
-###########################################################################
-# 2) validate a successful mapreduce operation after the execution
-###########################################################################
-        o,c= self.master_unit.run("su hdfs -c 'hdfs dfs -ls /user/ubuntu/terasortout/_SUCCESS'")
-        if c == 0:
-            amulet.raise_status(amulet.PASS, msg="mapreduce operation was success")
-        else:
-            amulet.raise_status(amulet.FAIL, msg="mapreduce operation was not a success")
-###########################################################################
-# validate a successful deletion of mapreduce operation result from hdfs
-###########################################################################
-    def test_hdfs_delete_dir(self):
-        o,c= self.master_unit.run("su hdfs -c 'hdfs dfs -rm -r /user/ubuntu/teragenout'")
-        if c == 0:
-            amulet.raise_status(amulet.PASS, msg="mapreduce result deleted from hdfs cluster")
-        else:
-            amulet.raise_status(amulet.FAIL, msg="mapreduce was not deleted from hdfs cluster")
+        """
+        Validate hadoop services on compute node have been started
+        """
+        output, retcode = self.compute_unit.run("pgrep -a java")
+        assert 'NodeManager' in output, "NodeManager not started"
+        assert 'DataNode' in output, "DataServer not started"
+
+    def test_hdfs_dir(self):
+        """
+        Validate admin few hadoop activities on HDFS cluster.
+            1) This test validates mkdir on hdfs cluster
+            2) This test validates change hdfs dir owner on the cluster
+            3) This test validates setting hdfs directory access permission on the cluster
+
+        NB: These are order-dependent, so must be done as part of a single test case.
+        """
+        output, retcode = self.master_unit.run("su hdfs -c 'hdfs dfs -mkdir -p /user/ubuntu'")
+        assert retcode == 0, "Created a user directory on hdfs FAILED:\n{}".format(output)
+        output, retcode = self.master_unit.run("su hdfs -c 'hdfs dfs -chown ubuntu:ubuntu /user/ubuntu'")
+        assert retcode == 0, "Assigning an owner to hdfs directory FAILED:\n{}".format(output)
+        output, retcode = self.master_unit.run("su hdfs -c 'hdfs dfs -chmod -R 755 /user/ubuntu'")
+        assert retcode == 0, "seting directory permission on hdfs FAILED:\n{}".format(output)
+
+    def test_yarn_mapreduce_exe(self):
+        """
+        Validate yarn mapreduce operations:
+            1) validate mapreduce execution - writing to hdfs
+            2) validate successful mapreduce operation after the execution
+            3) validate mapreduce execution - reading and writing to hdfs
+            4) validate successful mapreduce operation after the execution
+            5) validate successful deletion of mapreduce operation result from hdfs
+
+        NB: These are order-dependent, so must be done as part of a single test case.
+        """
+        jar_file = '/usr/hdp/current/hadoop-mapreduce-client/hadoop-mapreduce-examples.jar'
+        test_steps = [
+            ('teragen',      "su ubuntu -c 'hadoop jar {} teragen  10000 /user/ubuntu/teragenout'".format(jar_file)),
+            ('mapreduce #1', "su hdfs -c 'hdfs dfs -ls /user/ubuntu/teragenout/_SUCCESS'"),
+            ('terasort',     "su ubuntu -c 'hadoop jar {} terasort /user/ubuntu/teragenout /user/ubuntu/terasortout'".format(jar_file)),
+            ('mapreduce #2', "su hdfs -c 'hdfs dfs -ls /user/ubuntu/terasortout/_SUCCESS'"),
+            ('cleanup',      "su hdfs -c 'hdfs dfs -rm -r /user/ubuntu/teragenout'"),
+        ]
+        for name, step in test_steps:
+            output, retcode = self.master_unit.run(step)
+            assert retcode == 0, "{} FAILED:\n{}".format(name, output)
 
 
 if __name__ == '__main__':
-    runner = TestDeployment()
-    runner.run()
+    unittest.main()