AMBARI-3504. Add Resource Management library to project and rpms. (Andrew Onischuk via mahadev)

LICENSE.txt

@@ -286,3 +286,33 @@ A "contributor" is any person that distributes its contribution under this licen
 (D) If you distribute any portion of the software in source code form, you may do so only under this license by including a complete copy of this license with your distribution. If you distribute any portion of the software in compiled or object code form, you may only do so under a license that complies with this license.
 (E) The software is licensed "as-is." You bear the risk of using it. The contributors give no express warranties, guarantees or conditions. You may have additional consumer rights under your local laws which this license cannot change. To the extent permitted under your local laws, the contributors exclude the implied warranties of merchantability, fitness for a particular purpose and non-infringement.
 
+For Kokki, resource management library derived from:
+
+Copyright (c) 2009 Samuel Stauffer <samuel@descolada.com>
+All rights reserved.
+
+Redistribution and use in source and binary forms, with or without modification,
+are permitted provided that the following conditions are met:
+
+    1. Redistributions of source code must retain the above copyright notice, 
+       this list of conditions and the following disclaimer.
+    
+    2. Redistributions in binary form must reproduce the above copyright 
+       notice, this list of conditions and the following disclaimer in the
+       documentation and/or other materials provided with the distribution.
+
+    3. Neither the name of Samuel Stauffer nor the names of its contributors may be used
+       to endorse or promote products derived from this software without
+       specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
+ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR
+ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON
+ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+

NOTICE.txt

@@ -9,4 +9,8 @@ Component ambari-common/src/test/python are under the following copyright:
 Copyright (c) 2003-2012, Michael Foord
 All rights reserved.
 
-
+ 
+Resource management library derived from Kokki, which is under following copyright:
+ 
+Copyright (c) 2009 Samuel Stauffer <samuel@descolada.com>
+All rights reserved.

ambari-agent/pom.xml

@@ -39,7 +39,8 @@
     <skipTests>false</skipTests>
     <facter.tar>http://downloads.puppetlabs.com/facter/facter-1.6.10.tar.gz</facter.tar>
     <puppet.tar>http://downloads.puppetlabs.com/puppet/puppet-2.7.9.tar.gz</puppet.tar>
-    <install.dir>/usr/lib/python2.6/site-packages/ambari_agent</install.dir>
+    <agent.install.dir>/usr/lib/python2.6/site-packages/ambari_agent</agent.install.dir>
+    <resmgmt.install.dir>/usr/lib/python2.6/site-packages/resource_management</resmgmt.install.dir>
     <ruby.tar>http://public-repo-1.hortonworks.com/HDP-UTILS-1.1.0.15/repos/centos6/ruby-1.8.7-p370.tar.gz</ruby.tar>
     <lib.dir>/usr/lib/ambari-agent/lib</lib.dir>
     <python.ver>python &gt;= 2.6</python.ver>
@@ -122,7 +123,7 @@
                 <argument>unitTests.py</argument>
               </arguments>
               <environmentVariables>
-                <PYTHONPATH>${project.basedir}/../ambari-common/src/test/python:${project.basedir}/src/main/python/ambari_agent:${project.basedir}/src/main/puppet/modules/hdp-hadoop/files:$PYTHONPATH</PYTHONPATH>
+                <PYTHONPATH>${project.basedir}/../ambari-common/src/test/python:${project.basedir}/src/main/python/ambari_agent:${project.basedir}/src/main/python/resource_management:${project.basedir}/src/main/puppet/modules/hdp-hadoop/files:$PYTHONPATH</PYTHONPATH>
               </environmentVariables>
               <skip>${skipTests}</skip>
             </configuration>
@@ -192,13 +193,21 @@
           <autoRequires>false</autoRequires>
           <mappings>
             <mapping>
-              <directory>${install.dir}</directory>
+              <directory>${agent.install.dir}</directory>
               <sources>
                 <source>
                   <location>${project.build.directory}/${project.artifactId}-${project.version}/ambari_agent</location>
                 </source>
               </sources>
             </mapping>
+            <mapping>
+              <directory>${resmgmt.install.dir}</directory>
+              <sources>
+                <source>
+                  <location>${project.build.directory}/${project.artifactId}-${project.version}/resource_management</location>
+                </source>
+              </sources>
+            </mapping>
             <mapping>
               <directory>${lib.dir}</directory>
               <filemode>755</filemode>

ambari-agent/src/main/python/resource_management/__init__.py

@@ -0,0 +1,9 @@
+from resource_management.base import *
+from resource_management.environment import *
+from resource_management.exceptions import *
+from resource_management.providers import *
+from resource_management.resources import *
+from resource_management.source import *
+from resource_management.system import *
+
+__version__ = "0.4.1"

ambari-agent/src/main/python/resource_management/base.py

@@ -0,0 +1,205 @@
+#!/usr/bin/env python
+
+__all__ = ["Resource", "ResourceArgument", "ForcedListArgument",
+           "BooleanArgument"]
+
+import logging
+from resource_management.environment import Environment
+from resource_management.exceptions import Fail, InvalidArgument
+
+
+class ResourceArgument(object):
+  def __init__(self, default=None, required=False, allow_override=False):
+    self.required = False # Prevents the initial validate from failing
+    if hasattr(default, '__call__'):
+      self.default = default
+    else:
+      self.default = self.validate(default)
+    self.required = required
+    self.allow_override = allow_override
+
+  def validate(self, value):
+    if self.required and value is None:
+      raise InvalidArgument("Required argument %s missing" % self.name)
+    return value
+
+
+class ForcedListArgument(ResourceArgument):
+  def validate(self, value):
+    value = super(ForcedListArgument, self).validate(value)
+    if not isinstance(value, (tuple, list)):
+      value = [value]
+    return value
+
+
+class BooleanArgument(ResourceArgument):
+  def validate(self, value):
+    value = super(BooleanArgument, self).validate(value)
+    if not value in (True, False):
+      raise InvalidArgument(
+        "Expected a boolean for %s received %r" % (self.name, value))
+    return value
+
+
+class Accessor(object):
+  def __init__(self, name):
+    self.name = name
+
+  def __get__(self, obj, cls):
+    try:
+      return obj.arguments[self.name]
+    except KeyError:
+      val = obj._arguments[self.name].default
+      if hasattr(val, '__call__'):
+        val = val(obj)
+      return val
+
+  def __set__(self, obj, value):
+    obj.arguments[self.name] = obj._arguments[self.name].validate(value)
+
+
+class ResourceMetaclass(type):
+  # def __new__(cls, name, bases, attrs):
+  #     super_new = super(ResourceMetaclass, cls).__new__
+  #     return super_new(cls, name, bases, attrs)
+
+  def __init__(mcs, _name, bases, attrs):
+    mcs._arguments = getattr(bases[0], '_arguments', {}).copy()
+    for key, value in list(attrs.items()):
+      if isinstance(value, ResourceArgument):
+        value.name = key
+        mcs._arguments[key] = value
+        setattr(mcs, key, Accessor(key))
+
+
+class Resource(object):
+  __metaclass__ = ResourceMetaclass
+
+  is_updated = False
+
+  action = ForcedListArgument(default="nothing")
+  ignore_failures = BooleanArgument(default=False)
+  notifies = ResourceArgument(default=[])
+  subscribes = ResourceArgument(default=[])
+  not_if = ResourceArgument()
+  only_if = ResourceArgument()
+
+  actions = ["nothing"]
+
+  def __new__(cls, name, env=None, provider=None, **kwargs):
+    env = env or Environment.get_instance()
+    provider = provider or getattr(cls, 'provider', None)
+
+    r_type = cls.__name__
+    if r_type not in env.resources:
+      env.resources[r_type] = {}
+    if name not in env.resources[r_type]:
+      obj = super(Resource, cls).__new__(cls)
+      env.resources[r_type][name] = obj
+      env.resource_list.append(obj)
+      return obj
+
+    obj = env.resources[r_type][name]
+    if obj.provider != provider:
+      raise Fail("Duplicate resource %r with a different provider %r != %r" % (
+      obj, provider, obj.provider))
+
+    obj.override(**kwargs)
+    return obj
+
+  def __init__(self, name, env=None, provider=None, **kwargs):
+    if hasattr(self, 'name'):
+      return
+
+    self.name = name
+    self.env = env or Environment.get_instance()
+    self.provider = provider or getattr(self, 'provider', None)
+    self.log = logging.getLogger("resource_management.resource")
+
+    self.arguments = {}
+    for key, value in kwargs.items():
+      try:
+        arg = self._arguments[key]
+      except KeyError:
+        raise Fail("%s received unsupported argument %s" % (self, key))
+      else:
+        try:
+          self.arguments[key] = arg.validate(value)
+        except InvalidArgument, exc:
+          raise InvalidArgument("%s %s" % (self, exc))
+
+    self.log.debug("New resource %s: %s" % (self, self.arguments))
+    self.subscriptions = {'immediate': set(), 'delayed': set()}
+
+    for sub in self.subscribes:
+      if len(sub) == 2:
+        action, res = sub
+        immediate = False
+      else:
+        action, res, immediate = sub
+
+      res.subscribe(action, self, immediate)
+
+    for sub in self.notifies:
+      self.subscribe(*sub)
+
+    self.validate()
+
+  def validate(self):
+    pass
+
+  def subscribe(self, action, resource, immediate=False):
+    imm = "immediate" if immediate else "delayed"
+    sub = (action, resource)
+    self.subscriptions[imm].add(sub)
+
+  def updated(self):
+    self.is_updated = True
+
+  def override(self, **kwargs):
+    for key, value in kwargs.items():
+      try:
+        arg = self._arguments[key]
+      except KeyError:
+        raise Fail("%s received unsupported argument %s" % (self, key))
+      else:
+        if value != self.arguments.get(key):
+          if not arg.allow_override:
+            raise Fail(
+              "%s doesn't allow overriding argument '%s'" % (self, key))
+
+          try:
+            self.arguments[key] = arg.validate(value)
+          except InvalidArgument, exc:
+            raise InvalidArgument("%s %s" % (self, exc))
+    self.validate()
+
+  def __repr__(self):
+    return "%s['%s']" % (self.__class__.__name__, self.name)
+
+  def __unicode__(self):
+    return u"%s['%s']" % (self.__class__.__name__, self.name)
+
+  def __getstate__(self):
+    return dict(
+      name=self.name,
+      provider=self.provider,
+      arguments=self.arguments,
+      subscriptions=self.subscriptions,
+      subscribes=self.subscribes,
+      notifies=self.notifies,
+      env=self.env,
+    )
+
+  def __setstate__(self, state):
+    self.name = state['name']
+    self.provider = state['provider']
+    self.arguments = state['arguments']
+    self.subscriptions = state['subscriptions']
+    self.subscribes = state['subscribes']
+    self.notifies = state['notifies']
+    self.env = state['env']
+
+    self.log = logging.getLogger("resource_management.resource")
+
+    self.validate()

ambari-agent/src/main/python/resource_management/environment.py

@@ -0,0 +1,137 @@
+#!/usr/bin/env python
+
+__all__ = ["Environment"]
+
+import logging
+import os
+import shutil
+import subprocess
+from datetime import datetime
+
+from resource_management.exceptions import Fail
+from resource_management.providers import find_provider
+from resource_management.utils import AttributeDictionary
+from resource_management.system import System
+
+
+class Environment(object):
+  _instances = []
+
+  def __init__(self):
+    self.log = logging.getLogger("resource_management")
+    self.reset()
+
+  def reset(self):
+    self.system = System.get_instance()
+    self.config = AttributeDictionary()
+    self.resources = {}
+    self.resource_list = []
+    self.delayed_actions = set()
+    self.update_config({
+      'date': datetime.now(),
+      'resource_management.backup.path': '/tmp/resource_management/backup',
+      'resource_management.backup.prefix': datetime.now().strftime("%Y%m%d%H%M%S"),
+    })
+
+  def backup_file(self, path):
+    if self.config.kokki.backup:
+      if not os.path.exists(self.config.kokki.backup.path):
+        os.makedirs(self.config.kokki.backup.path, 0700)
+      new_name = self.config.kokki.backup.prefix + path.replace('/', '-')
+      backup_path = os.path.join(self.config.kokki.backup.path, new_name)
+      self.log.info("backing up %s to %s" % (path, backup_path))
+      shutil.copy(path, backup_path)
+
+  def update_config(self, attributes, overwrite=True):
+    for key, value in attributes.items():
+      attr = self.config
+      path = key.split('.')
+      for pth in path[:-1]:
+        if pth not in attr:
+          attr[pth] = AttributeDictionary()
+        attr = attr[pth]
+      if overwrite or path[-1] not in attr:
+        attr[path[-1]] = value
+
+  def run_action(self, resource, action):
+    self.log.debug("Performing action %s on %s" % (action, resource))
+
+    provider_class = find_provider(self, resource.__class__.__name__,
+                                   resource.provider)
+    provider = provider_class(resource)
+    try:
+      provider_action = getattr(provider, 'action_%s' % action)
+    except AttributeError:
+      raise Fail("%r does not implement action %s" % (provider, action))
+    provider_action()
+
+    if resource.is_updated:
+      for action, res in resource.subscriptions['immediate']:
+        self.log.info(
+          "%s sending %s action to %s (immediate)" % (resource, action, res))
+        self.run_action(res, action)
+      for action, res in resource.subscriptions['delayed']:
+        self.log.info(
+          "%s sending %s action to %s (delayed)" % (resource, action, res))
+      self.delayed_actions |= resource.subscriptions['delayed']
+
+  def _check_condition(self, cond):
+    if hasattr(cond, '__call__'):
+      return cond()
+
+    if isinstance(cond, basestring):
+      ret = subprocess.call(cond, shell=True)
+      return ret == 0
+
+    raise Exception("Unknown condition type %r" % cond)
+
+  def run(self):
+    with self:
+      # Run resource actions
+      for resource in self.resource_list:
+        self.log.debug("Running resource %r" % resource)
+
+        if resource.not_if is not None and self._check_condition(
+          resource.not_if):
+          self.log.debug("Skipping %s due to not_if" % resource)
+          continue
+
+        if resource.only_if is not None and not self._check_condition(
+          resource.only_if):
+          self.log.debug("Skipping %s due to only_if" % resource)
+          continue
+
+        for action in resource.action:
+          self.run_action(resource, action)
+
+      # Run delayed actions
+      while self.delayed_actions:
+        action, resource = self.delayed_actions.pop()
+        self.run_action(resource, action)
+
+  @classmethod
+  def get_instance(cls):
+    return cls._instances[-1]
+
+  def __enter__(self):
+    self.__class__._instances.append(self)
+    return self
+
+  def __exit__(self, exc_type, exc_val, exc_tb):
+    self.__class__._instances.pop()
+    return False
+
+  def __getstate__(self):
+    return dict(
+      config=self.config,
+      resources=self.resources,
+      resource_list=self.resource_list,
+      delayed_actions=self.delayed_actions,
+    )
+
+  def __setstate__(self, state):
+    self.__init__()
+    self.config = state['config']
+    self.resources = state['resources']
+    self.resource_list = state['resource_list']
+    self.delayed_actions = state['delayed_actions']

ambari-agent/src/main/python/resource_management/exceptions.py

@@ -0,0 +1,6 @@
+class Fail(Exception):
+  pass
+
+
+class InvalidArgument(Fail):
+  pass

ambari-agent/src/main/python/resource_management/providers/__init__.py

@@ -0,0 +1,80 @@
+__all__ = ["Provider", "find_provider"]
+
+import logging
+from resource_management.exceptions import Fail
+
+
+class Provider(object):
+  def __init__(self, resource):
+    self.log = logging.getLogger("resource_management.provider")
+    self.resource = resource
+
+  def action_nothing(self):
+    pass
+
+  def __repr__(self):
+    return self.__unicode__()
+
+  def __unicode__(self):
+    return u"%s[%s]" % (self.__class__.__name__, self.resource)
+
+
+PROVIDERS = dict(
+  debian=dict(
+    Package="resource_management.providers.package.apt.DebianAptProvider",
+    Service="resource_management.providers.service.debian.DebianServiceProvider",
+  ),
+  ubuntu=dict(
+    Package="resource_management.providers.package.apt.DebianAptProvider",
+    Service="resource_management.providers.service.debian.DebianServiceProvider",
+  ),
+  redhat=dict(
+    Service="resource_management.providers.service.redhat.RedhatServiceProvider",
+    Package="resource_management.providers.package.yumrpm.YumProvider",
+  ),
+  centos=dict(
+    Service="resource_management.providers.service.redhat.RedhatServiceProvider",
+    Package="resource_management.providers.package.yumrpm.YumProvider",
+  ),
+  fedora=dict(
+    Service="resource_management.providers.service.redhat.RedhatServiceProvider",
+    Package="resource_management.providers.package.yumrpm.YumProvider",
+  ),
+  amazon=dict(
+    Service="resource_management.providers.service.redhat.RedhatServiceProvider",
+    Package="resource_management.providers.package.yumrpm.YumProvider",
+  ),
+  gentoo=dict(
+    Package="resource_management.providers.package.emerge.GentooEmergeProvider",
+    Service="resource_management.providers.service.gentoo.GentooServiceProvider",
+  ),
+  default=dict(
+    File="resource_management.providers.system.FileProvider",
+    Directory="resource_management.providers.system.DirectoryProvider",
+    Link="resource_management.providers.system.LinkProvider",
+    Execute="resource_management.providers.system.ExecuteProvider",
+    Script="resource_management.providers.system.ScriptProvider",
+    Mount="resource_management.providers.mount.MountProvider",
+    User="resource_management.providers.accounts.UserProvider",
+    Group="resource_management.providers.accounts.GroupProvider",
+  ),
+)
+
+
+def find_provider(env, resource, class_path=None):
+  if not class_path:
+    try:
+      class_path = PROVIDERS[env.system.platform][resource]
+    except KeyError:
+      class_path = PROVIDERS["default"][resource]
+
+  if class_path.startswith('*'):
+    cookbook, classname = class_path[1:].split('.')
+    return getattr(env.cookbooks[cookbook], classname)
+
+  try:
+    mod_path, class_name = class_path.rsplit('.', 1)
+  except ValueError:
+    raise Fail("Unable to find provider for %s as %s" % (resource, class_path))
+  mod = __import__(mod_path, {}, {}, [class_name])
+  return getattr(mod, class_name)

ambari-agent/src/main/python/resource_management/providers/accounts.py

@@ -0,0 +1,97 @@
+from __future__ import with_statement
+
+import grp
+import pwd
+import subprocess
+from resource_management.providers import Provider
+
+
+class UserProvider(Provider):
+  def action_create(self):
+    if not self.user:
+      command = ['useradd', "-m"]
+
+      useradd_options = dict(
+        comment="-c",
+        gid="-g",
+        uid="-u",
+        shell="-s",
+        password="-p",
+        home="-d",
+      )
+
+      if self.resource.system:
+        command.append("--system")
+
+      if self.resource.groups:
+        command += ["-G", ",".join(self.resource.groups)]
+
+      for option_name, option_flag in useradd_options.items():
+        option_value = getattr(self.resource, option_name)
+        if option_flag and option_value:
+          command += [option_flag, str(option_value)]
+
+      command.append(self.resource.username)
+
+      subprocess.check_call(command)
+      self.resource.updated()
+      self.log.info("Added user %s" % self.resource)
+
+  def action_remove(self):
+    if self.user:
+      command = ['userdel', self.resource.username]
+      subprocess.check_call(command)
+      self.resource.updated()
+      self.log.info("Removed user %s" % self.resource)
+
+  @property
+  def user(self):
+    try:
+      return pwd.getpwnam(self.resource.username)
+    except KeyError:
+      return None
+
+
+class GroupProvider(Provider):
+  def action_create(self):
+    group = self.group
+    if not group:
+      command = ['groupadd']
+
+      groupadd_options = dict(
+        gid="-g",
+        password="-p",
+      )
+
+      for option_name, option_flag in groupadd_options.items():
+        option_value = getattr(self.resource, option_name)
+        if option_flag and option_value:
+          command += [option_flag, str(option_value)]
+
+      command.append(self.resource.group_name)
+
+      subprocess.check_call(command)
+      self.resource.updated()
+      self.log.info("Added group %s" % self.resource)
+
+      group = self.group
+
+      # if self.resource.members is not None:
+      #     current_members = set(group.gr_mem)
+      #     members = set(self.resource.members)
+      #     for u in current_members - members:
+      #         pass
+
+  def action_remove(self):
+    if self.user:
+      command = ['groupdel', self.resource.group_name]
+      subprocess.check_call(command)
+      self.resource.updated()
+      self.log.info("Removed group %s" % self.resource)
+
+  @property
+  def group(self):
+    try:
+      return grp.getgrnam(self.resource.group_name)
+    except KeyError:
+      return None

ambari-agent/src/main/python/resource_management/providers/mount.py

@@ -0,0 +1,118 @@
+from __future__ import with_statement
+
+import os
+import re
+from subprocess import Popen, PIPE, STDOUT, check_call
+from resource_management.base import Fail
+from resource_management.providers import Provider
+
+
+class MountProvider(Provider):
+  def action_mount(self):
+    if not os.path.exists(self.resource.mount_point):
+      os.makedirs(self.resource.mount_point)
+
+    if self.is_mounted():
+      self.log.debug("%s already mounted" % self)
+    else:
+      args = ["mount"]
+      if self.resource.fstype:
+        args += ["-t", self.resource.fstype]
+      if self.resource.options:
+        args += ["-o", ",".join(self.resource.options)]
+      if self.resource.device:
+        args.append(self.resource.device)
+      args.append(self.resource.mount_point)
+
+      check_call(args)
+
+      self.log.info("%s mounted" % self)
+      self.resource.updated()
+
+  def action_umount(self):
+    if self.is_mounted():
+      check_call(["umount", self.resource.mount_point])
+
+      self.log.info("%s unmounted" % self)
+      self.resource.updated()
+    else:
+      self.log.debug("%s is not mounted" % self)
+
+  def action_enable(self):
+    if self.is_enabled():
+      self.log.debug("%s already enabled" % self)
+    else:
+      if not self.resource.device:
+        raise Fail("[%s] device not set but required for enable action" % self)
+      if not self.resource.fstype:
+        raise Fail("[%s] fstype not set but required for enable action" % self)
+
+      with open("/etc/fstab", "a") as fp:
+        fp.write("%s %s %s %s %d %d\n" % (
+          self.resource.device,
+          self.resource.mount_point,
+          self.resource.fstype,
+          ",".join(self.resource.options or ["defaults"]),
+          self.resource.dump,
+          self.resource.passno,
+        ))
+
+      self.log.info("%s enabled" % self)
+      self.resource.updated()
+
+  def action_disable(self):
+    pass # TODO
+
+  def is_mounted(self):
+    if not os.path.exists(self.resource.mount_point):
+      return False
+
+    if self.resource.device and not os.path.exists(self.resource.device):
+      raise Fail("%s Device %s does not exist" % (self, self.resource.device))
+
+    mounts = self.get_mounted()
+    for m in mounts:
+      if m['mount_point'] == self.resource.mount_point:
+        return True
+
+    return False
+
+  def is_enabled(self):
+    mounts = self.get_fstab()
+    for m in mounts:
+      if m['mount_point'] == self.resource.mount_point:
+        return True
+
+    return False
+
+  def get_mounted(self):
+    p = Popen("mount", stdout=PIPE, stderr=STDOUT, shell=True)
+    out = p.communicate()[0]
+    if p.wait() != 0:
+      raise Fail("[%s] Getting list of mounts (calling mount) failed" % self)
+
+    mounts = [x.split(' ') for x in out.strip().split('\n')]
+
+    return [dict(
+      device=m[0],
+      mount_point=m[2],
+      fstype=m[4],
+      options=m[5][1:-1].split(','),
+    ) for m in mounts if m[1] == "on" and m[3] == "type"]
+
+  def get_fstab(self):
+    mounts = []
+    with open("/etc/fstab", "r") as fp:
+      for line in fp:
+        line = line.split('#', 1)[0].strip()
+        mount = re.split('\s+', line)
+        if len(mount) == 6:
+          mounts.append(dict(
+            device=mount[0],
+            mount_point=mount[1],
+            fstype=mount[2],
+            options=mount[3].split(","),
+            dump=int(mount[4]),
+            passno=int(mount[5]),
+          ))
+    return mounts

ambari-agent/src/main/python/resource_management/providers/package/__init__.py

@@ -0,0 +1,69 @@
+from resource_management.base import Fail
+from resource_management.providers import Provider
+
+
+class PackageProvider(Provider):
+  def __init__(self, *args, **kwargs):
+    super(PackageProvider, self).__init__(*args, **kwargs)
+    self.get_current_status()
+
+  def get_current_status(self):
+    raise NotImplementedError()
+
+  def install_package(self, name, version):
+    raise NotImplementedError()
+
+  def remove_package(self, name):
+    raise NotImplementedError()
+
+  def purge_package(self, name):
+    raise NotImplementedError()
+
+  def upgrade_package(self, name, version):
+    raise NotImplementedError()
+
+  def action_install(self):
+    if self.resource.version != None and self.resource.version != self.current_version:
+      install_version = self.resource.version
+    elif self.current_version is None:
+      install_version = self.candidate_version
+    else:
+      return
+
+    if not install_version:
+      raise Fail(
+        "No version specified, and no candidate version available for package %s." % self.resource.package_name)
+
+    self.log.info(
+      "Install %s version %s (resource %s, current %s, candidate %s) location %s",
+      self.resource.package_name, install_version, self.resource.version,
+      self.current_version, self.candidate_version, self.resource.location)
+
+    status = self.install_package(self.resource.location, install_version)
+    if status:
+      self.resource.updated()
+
+  def action_upgrade(self):
+    if self.current_version != self.candidate_version:
+      orig_version = self.current_version or "uninstalled"
+      self.log.info("Upgrading %s from version %s to %s",
+                    str(self.resource), orig_version, self.candidate_version)
+
+      status = self.upgrade_package(self.resource.location,
+                                    self.candidate_version)
+      if status:
+        self.resource.updated()
+
+  def action_remove(self):
+    if self.current_version:
+      self.log.info("Remove %s version %s", self.resource.package_name,
+                    self.current_version)
+      self.remove_package(self.resource.package_name)
+      self.resource.updated()
+
+  def action_purge(self):
+    if self.current_version:
+      self.log.info("Purging %s version %s", self.resource.package_name,
+                    self.current_version)
+      self.purge_package(self.resource.package_name)
+      self.resource.updated()

ambari-agent/src/main/python/resource_management/providers/package/apt.py

@@ -0,0 +1,96 @@
+import glob
+import os
+import shutil
+import tempfile
+from subprocess import Popen, STDOUT, PIPE, check_call, CalledProcessError
+from resource_management.base import Fail
+from resource_management.providers.package import PackageProvider
+
+
+class DebianAptProvider(PackageProvider):
+  def get_current_status(self):
+    self.current_version = None
+    self.candidate_version = None
+
+    proc = Popen("apt-cache policy %s" % self.resource.package_name, shell=True,
+                 stdout=PIPE)
+    out = proc.communicate()[0]
+    for line in out.split("\n"):
+      line = line.strip().split(':', 1)
+      if len(line) != 2:
+        continue
+
+      ver = line[1].strip()
+      if line[0] == "Installed":
+        self.current_version = None if ver == '(none)' else ver
+        self.log.debug("Current version of package %s is %s" % (
+        self.resource.package_name, self.current_version))
+      elif line[0] == "Candidate":
+        self.candidate_version = ver
+
+    if self.candidate_version == "(none)":
+      raise Fail(
+        "APT does not provide a version of package %s" % self.resource.package_name)
+
+  def install_package(self, name, version):
+    if self.resource.build_vars:
+      self._install_package_source(name, version)
+    else:
+      self._install_package_default(name, version)
+
+  def _install_package_default(self, name, version):
+    return 0 == check_call(
+      "DEBIAN_FRONTEND=noninteractive apt-get -q -y install %s=%s" % (
+      name, version),
+      shell=True, stdout=PIPE, stderr=STDOUT)
+
+  def _install_package_source(self, name, version):
+    build_vars = " ".join(self.resource.build_vars)
+    run_check_call = lambda s, **kw: check_call(s, shell=True, stdout=PIPE,
+                                                stderr=STDOUT, **kw)
+    pkgdir = tempfile.mkdtemp(suffix=name)
+
+    try:
+      run_check_call(
+        "DEBIAN_FRONTEND=noninteractive apt-get -q -y install fakeroot")
+      run_check_call(
+        "DEBIAN_FRONTEND=noninteractive apt-get -q -y build-dep %s=%s" % (
+        name, version))
+      run_check_call(
+        "DEBIAN_FRONTEND=noninteractive apt-get -q -y source %s=%s" % (
+        name, version), cwd=pkgdir)
+
+      try:
+        builddir =
+        [p for p in glob.iglob("%s/%s*" % (pkgdir, name)) if os.path.isdir(p)][
+          0]
+      except IndexError:
+        raise Fail(
+          "Couldn't install %s from source: apt-get source created an unfamiliar directory structure." % name)
+
+      run_check_call("%s fakeroot debian/rules binary > /dev/null" % build_vars,
+                     cwd=builddir)
+
+      # NOTE: I can't figure out why this call returns non-zero sometimes, though everything seems to work.
+      # Just ignoring checking for now.
+      try:
+        run_check_call("dpkg -i *.deb > /dev/null", cwd=pkgdir)
+      except CalledProcessError:
+        pass
+    finally:
+      shutil.rmtree(pkgdir)
+
+    return True
+
+  def remove_package(self, name):
+    return 0 == check_call(
+      "DEBIAN_FRONTEND=noninteractive apt-get -q -y remove %s" % name,
+      shell=True, stdout=PIPE, stderr=STDOUT)
+
+  def purge_package(self, name):
+    return 0 == check_call(
+      "DEBIAN_FRONTEND=noninteractive apt-get -q -y purge %s" % name,
+      shell=True, stdout=PIPE, stderr=STDOUT)
+
+  def upgrade_package(self, name, version):
+    return self.install_package(name, version)

ambari-agent/src/main/python/resource_management/providers/package/easy_install.py

@@ -0,0 +1,61 @@
+import re
+from subprocess import check_call, Popen, PIPE, STDOUT
+from resource_management.providers.package import PackageProvider
+
+VERSION_RE = re.compile(r'\S\S(.*)\/(.*)-(.*)-py(.*).egg\S')
+BEST_MATCH_RE = re.compile(r'Best match: (.*) (.*)\n')
+
+
+class EasyInstallProvider(PackageProvider):
+  def get_current_status(self):
+    proc = Popen(["python", "-c", "import %s; print %s.__path__" % (
+    self.resource.package_name, self.resource.package_name)], stdout=PIPE,
+                 stderr=STDOUT)
+    path = proc.communicate()[0]
+    if proc.wait() != 0:
+      self.current_version = None
+    else:
+      match = VERSION_RE.search(path)
+      if match:
+        self.current_version = match.group(3)
+      else:
+        self.current_version = "unknown"
+
+  @property
+  def candidate_version(self):
+    if not hasattr(self, '_candidate_version'):
+      proc = Popen(
+        [self.easy_install_binary_path, "-n", self.resource.package_name],
+        stdout=PIPE, stderr=STDOUT)
+      out = proc.communicate()[0]
+      res = proc.wait()
+      if res != 0:
+        self.log.warning(
+          "easy_install check returned a non-zero result (%d) %s" % (
+          res, self.resource))
+        #     self._candidate_version = None
+      # else:
+      match = BEST_MATCH_RE.search(out)
+      if not match:
+        self._candidate_version = None
+      else:
+        self._candidate_version = match.group(2)
+    return self._candidate_version
+
+  @property
+  def easy_install_binary_path(self):
+    return "easy_install"
+
+  def install_package(self, name, version):
+    check_call(
+      [self.easy_install_binary_path, "-U", "%s==%s" % (name, version)],
+      stdout=PIPE, stderr=STDOUT)
+
+  def upgrade_package(self, name, version):
+    self.install_package(name, version)
+
+  def remove_package(self, name):
+    check_call([self.easy_install_binary_path, "-m", name])
+
+  def purge_package(self, name):
+    self.remove_package(name)

ambari-agent/src/main/python/resource_management/providers/package/emerge.py

@@ -0,0 +1,49 @@
+from subprocess import Popen, STDOUT, PIPE, check_call
+from resource_management.base import Fail
+from resource_management.providers.package import PackageProvider
+
+
+class GentooEmergeProvider(PackageProvider):
+  def get_current_status(self):
+    self.current_version = None
+    self.candidate_version = None
+
+    proc = Popen("qlist --installed --exact --verbose --nocolor %s"
+                 % self.resource.package_name, shell=True, stdout=PIPE)
+    out = proc.communicate()[0]
+    for line in out.split("\n"):
+      line = line.split('/', 1)
+      if len(line) != 2:
+        continue
+      _category, nameversion = line
+      _name, version = nameversion.split('-', 1)
+      self.current_version = version
+      self.log.debug("Current version of package %s is %s",
+                     self.resource.package_name, self.current_version)
+
+    proc = Popen(
+      "emerge --pretend --quiet --color n %s" % self.resource.package_name,
+      shell=True, stdout=PIPE)
+    out = proc.communicate()[0]
+    for line in out.split("\n"):
+      line = line.strip(' [').split(']', 1)
+      if len(line) != 2:
+        continue
+
+      # kind, flag = line[0].split()
+      _category, nameversion = line[1].split('/', 1)
+      _name, version = nameversion.split('-', 1)
+      self.candidate_version = version
+      self.log.debug("Candidate version of package %s is %s",
+                     self.resource.package_name, self.candidate_version)
+
+    if self.candidate_version is None:
+      raise Fail(
+        "emerge does not provide a version of package %s" % self.resource.package_name)
+
+  def install_package(self, name, version):
+    return 0 == check_call("emerge --color n =%s-%s" % (name, version),
+                           shell=True, stdout=PIPE, stderr=STDOUT)
+
+  def upgrade_package(self, name, version):
+    return self.install_package(name, version)

ambari-agent/src/main/python/resource_management/providers/package/yumrpm.py

@@ -0,0 +1,46 @@
+from resource_management.providers.package import PackageProvider
+import yum
+
+
+class DummyCallback(object):
+  def event(self, state, data=None):
+    pass
+
+
+class YumProvider(PackageProvider):
+  def get_current_status(self):
+    self.candidate_version = None
+    self.current_version = None
+    yb = yum.YumBase()
+    yb.doConfigSetup()
+    yb.doTsSetup()
+    yb.doRpmDBSetup()
+    for pkg in yb.rpmdb.returnPackages():
+      if pkg.name == self.resource.package_name:
+        self.current_version = pkg.version
+        self.log.debug("Current version of %s is %s" % (
+        self.resource.package_name, self.current_version))
+    searchlist = ['name', 'version']
+    args = [self.resource.package_name]
+    matching = yb.searchPackages(searchlist, args)
+    for po in matching:
+      if po.name == self.resource.package_name:
+        self.candidate_version = po.version
+        self.log.debug("Candidate version of %s is %s" % (
+        self.resource.package_name, self.current_version))
+
+  def install_package(self, name, version):
+    yb = yum.YumBase()
+    yb.doGenericSetup()
+    yb.doRepoSetup()
+    #TODO: Handle locks not being available
+    yb.doLock()
+    yb.install(pattern=name)
+    yb.buildTransaction()
+    #yb.conf.setattr('assumeyes',True)
+    yb.processTransaction(callback=DummyCallback())
+    yb.closeRpmDB()
+    yb.doUnlock()
+
+  def upgrade_package(self, name, version):
+    return self.install_package(name, version)

ambari-agent/src/main/python/resource_management/providers/service/__init__.py

@@ -0,0 +1,86 @@
+import os
+import subprocess
+
+from resource_management.base import Fail
+from resource_management.providers import Provider
+
+
+class ServiceProvider(Provider):
+  def action_start(self):
+    if not self.status():
+      self._exec_cmd("start", 0)
+      self.resource.updated()
+
+  def action_stop(self):
+    if self.status():
+      self._exec_cmd("stop", 0)
+      self.resource.updated()
+
+  def action_restart(self):
+    if not self.status():
+      self._exec_cmd("start", 0)
+      self.resource.updated()
+    else:
+      self._exec_cmd("restart", 0)
+      self.resource.updated()
+
+  def action_reload(self):
+    if not self.status():
+      self._exec_cmd("start", 0)
+      self.resource.updated()
+    else:
+      self._exec_cmd("reload", 0)
+      self.resource.updated()
+
+  def status(self):
+    return self._exec_cmd("status") == 0
+
+  def _exec_cmd(self, command, expect=None):
+    if command != "status":
+      self.log.info("%s command '%s'" % (self.resource, command))
+
+    custom_cmd = getattr(self.resource, "%s_command" % command, None)
+    if custom_cmd:
+      self.log.debug("%s executing '%s'" % (self.resource, custom_cmd))
+      if hasattr(custom_cmd, "__call__"):
+        if custom_cmd():
+          ret = 0
+        else:
+          ret = 1
+      else:
+        ret = subprocess.call(custom_cmd, shell=True,
+                              stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+    else:
+      ret = self._init_cmd(command)
+
+    if expect is not None and expect != ret:
+      raise Fail("%r command %s for service %s failed" % (
+      self, command, self.resource.service_name))
+    return ret
+
+  def _init_cmd(self, command):
+    if self._upstart:
+      if command == "status":
+        proc = subprocess.Popen(
+          ["/sbin/" + command, self.resource.service_name],
+          stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+        out = proc.communicate()[0]
+        _proc, state = out.strip().split(' ', 1)
+        ret = 0 if state != "stop/waiting" else 1
+      else:
+        ret = subprocess.call(["/sbin/" + command, self.resource.service_name],
+                              stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+    else:
+      ret = subprocess.call(
+        ["/etc/init.d/%s" % self.resource.service_name, command],
+        stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+    return ret
+
+  @property
+  def _upstart(self):
+    try:
+      return self.__upstart
+    except AttributeError:
+      self.__upstart = os.path.exists("/sbin/start") \
+        and os.path.exists("/etc/init/%s.conf" % self.resource.service_name)
+    return self.__upstart

ambari-agent/src/main/python/resource_management/providers/service/debian.py

@@ -0,0 +1,8 @@
+__all__ = ["DebianServiceProvider"]
+
+from resource_management.providers.service import ServiceProvider
+
+
+class DebianServiceProvider(ServiceProvider):
+  def enable_runlevel(self, runlevel):
+    pass

ambari-agent/src/main/python/resource_management/providers/service/gentoo.py

@@ -0,0 +1,8 @@
+__all__ = ["GentooServiceProvider"]
+
+from resource_management.providers.service import ServiceProvider
+
+
+class GentooServiceProvider(ServiceProvider):
+  def enable_runlevel(self, runlevel):
+    pass

ambari-agent/src/main/python/resource_management/providers/service/redhat.py

@@ -0,0 +1,8 @@
+__all__ = ["RedhatServiceProvider"]
+
+from resource_management.providers.service import ServiceProvider
+
+
+class RedhatServiceProvider(ServiceProvider):
+  def enable_runlevel(self, runlevel):
+    pass

ambari-agent/src/main/python/resource_management/providers/system.py

@@ -0,0 +1,210 @@
+from __future__ import with_statement
+
+import grp
+import os
+import pwd
+import subprocess
+from resource_management.base import Fail
+from resource_management.providers import Provider
+
+
+def _coerce_uid(user):
+  try:
+    uid = int(user)
+  except ValueError:
+    uid = pwd.getpwnam(user).pw_uid
+  return uid
+
+
+def _coerce_gid(group):
+  try:
+    gid = int(group)
+  except ValueError:
+    gid = grp.getgrnam(group).gr_gid
+  return gid
+
+
+def _ensure_metadata(path, user, group, mode=None, log=None):
+  stat = os.stat(path)
+  updated = False
+
+  if mode:
+    existing_mode = stat.st_mode & 07777
+    if existing_mode != mode:
+      log and log.info("Changing permission for %s from %o to %o" % (
+      path, existing_mode, mode))
+      os.chmod(path, mode)
+      updated = True
+
+  if user:
+    uid = _coerce_uid(user)
+    if stat.st_uid != uid:
+      log and log.info(
+        "Changing owner for %s from %d to %s" % (path, stat.st_uid, user))
+      os.chown(path, uid, -1)
+      updated = True
+
+  if group:
+    gid = _coerce_gid(group)
+    if stat.st_gid != gid:
+      log and log.info(
+        "Changing group for %s from %d to %s" % (path, stat.st_gid, group))
+      os.chown(path, -1, gid)
+      updated = True
+
+  return updated
+
+
+class FileProvider(Provider):
+  def action_create(self):
+    path = self.resource.path
+    write = False
+    content = self._get_content()
+    if not os.path.exists(path):
+      write = True
+      reason = "it doesn't exist"
+    else:
+      if content is not None:
+        with open(path, "rb") as fp:
+          old_content = fp.read()
+        if content != old_content:
+          write = True
+          reason = "contents don't match"
+          self.resource.env.backup_file(path)
+
+    if write:
+      self.log.info("Writing %s because %s" % (self.resource, reason))
+      with open(path, "wb") as fp:
+        if content:
+          fp.write(content)
+      self.resource.updated()
+
+    if _ensure_metadata(self.resource.path, self.resource.owner,
+                        self.resource.group, mode=self.resource.mode,
+                        log=self.log):
+      self.resource.updated()
+
+  def action_delete(self):
+    path = self.resource.path
+    if os.path.exists(path):
+      self.log.info("Deleting %s" % self.resource)
+      os.unlink(path)
+      self.resource.updated()
+
+  def action_touch(self):
+    path = self.resource.path
+    with open(path, "a"):
+      pass
+
+  def _get_content(self):
+    content = self.resource.content
+    if content is None:
+      return None
+    elif isinstance(content, basestring):
+      return content
+    elif hasattr(content, "__call__"):
+      return content()
+    raise Fail("Unknown source type for %s: %r" % (self, content))
+
+
+class DirectoryProvider(Provider):
+  def action_create(self):
+    path = self.resource.path
+    if not os.path.exists(path):
+      self.log.info("Creating directory %s" % self.resource)
+      if self.resource.recursive:
+        os.makedirs(path, self.resource.mode or 0755)
+      else:
+        os.mkdir(path, self.resource.mode or 0755)
+      self.resource.updated()
+
+    if _ensure_metadata(path, self.resource.owner, self.resource.group,
+                        mode=self.resource.mode, log=self.log):
+      self.resource.updated()
+
+  def action_delete(self):
+    path = self.resource.path
+    if os.path.exists(path):
+      self.log.info("Removing directory %s" % self.resource)
+      os.rmdir(path)
+      # TODO: recursive
+      self.resource.updated()
+
+
+class LinkProvider(Provider):
+  def action_create(self):
+    path = self.resource.path
+
+    if os.path.lexists(path):
+      oldpath = os.path.realpath(path)
+      if oldpath == self.resource.to:
+        return
+      if not os.path.islink(path):
+        raise Fail(
+          "%s trying to create a symlink with the same name as an existing file or directory" % self)
+      self.log.info("%s replacing old symlink to %s" % (self, oldpath))
+      os.unlink(path)
+
+    if self.resource.hard:
+      self.log.info("Creating hard %s" % self.resource)
+      os.link(self.resource.to, path)
+      self.resource.updated()
+    else:
+      self.log.info("Creating symbolic %s" % self.resource)
+      os.symlink(self.resource.to, path)
+      self.resource.updated()
+
+  def action_delete(self):
+    path = self.resource.path
+    if os.path.exists(path):
+      self.log.info("Deleting %s" % self.resource)
+      os.unlink(path)
+      self.resource.updated()
+
+
+def _preexec_fn(resource):
+  def preexec():
+    if resource.group:
+      gid = _coerce_gid(resource.group)
+      os.setgid(gid)
+      os.setegid(gid)
+    if resource.user:
+      uid = _coerce_uid(resource.user)
+      os.setuid(uid)
+      os.seteuid(uid)
+
+  return preexec
+
+
+class ExecuteProvider(Provider):
+  def action_run(self):
+    if self.resource.creates:
+      if os.path.exists(self.resource.creates):
+        return
+
+    self.log.info("Executing %s" % self.resource)
+
+    ret = subprocess.call(self.resource.command, shell=True,
+                          cwd=self.resource.cwd, env=self.resource.environment,
+                          preexec_fn=_preexec_fn(self.resource))
+
+    if self.resource.returns and ret not in self.resource.returns:
+      raise Fail("%s failed, returned %d instead of %s" % (
+      self, ret, self.resource.returns))
+    self.resource.updated()
+
+
+class ScriptProvider(Provider):
+  def action_run(self):
+    from tempfile import NamedTemporaryFile
+
+    self.log.info("Running script %s" % self.resource)
+    with NamedTemporaryFile(prefix="resource_management-script", bufsize=0) as tf:
+      tf.write(self.resource.code)
+      tf.flush()
+
+      _ensure_metadata(tf.name, self.resource.user, self.resource.group)
+      subprocess.call([self.resource.interpreter, tf.name],
+                      cwd=self.resource.cwd, env=self.resource.environment,
+                      preexec_fn=_preexec_fn(self.resource))
+    self.resource.updated()

ambari-agent/src/main/python/resource_management/resources/__init__.py

@@ -0,0 +1,4 @@
+from resource_management.resources.accounts import *
+from resource_management.resources.packaging import *
+from resource_management.resources.service import *
+from resource_management.resources.system import *

ambari-agent/src/main/python/resource_management/resources/accounts.py

@@ -0,0 +1,31 @@
+__all__ = ["Group", "User"]
+
+from resource_management.base import Resource, ForcedListArgument, ResourceArgument, BooleanArgument
+
+
+class Group(Resource):
+  action = ForcedListArgument(default="create")
+  group_name = ResourceArgument(default=lambda obj: obj.name)
+  gid = ResourceArgument()
+  members = ForcedListArgument()
+  password = ResourceArgument()
+  # append = BooleanArgument(default=False) # NOT SUPPORTED
+
+  actions = Resource.actions + ["create", "remove", "modify", "manage", "lock",
+                                "unlock"]
+
+
+class User(Resource):
+  action = ForcedListArgument(default="create")
+  username = ResourceArgument(default=lambda obj: obj.name)
+  comment = ResourceArgument()
+  uid = ResourceArgument()
+  gid = ResourceArgument()
+  groups = ForcedListArgument() # supplementary groups
+  home = ResourceArgument()
+  shell = ResourceArgument(default="/bin/bash")
+  password = ResourceArgument()
+  system = BooleanArgument(default=False)
+
+  actions = Resource.actions + ["create", "remove", "modify", "manage", "lock",
+                                "unlock"]

ambari-agent/src/main/python/resource_management/resources/packaging.py

@@ -0,0 +1,12 @@
+__all__ = ["Package"]
+
+from resource_management.base import Resource, ForcedListArgument, ResourceArgument
+
+
+class Package(Resource):
+  action = ForcedListArgument(default="install")
+  package_name = ResourceArgument(default=lambda obj: obj.name)
+  location = ResourceArgument(default=lambda obj: obj.package_name)
+  version = ResourceArgument()
+  actions = ["install", "upgrade", "remove", "purge"]
+  build_vars = ForcedListArgument(default=[])

ambari-agent/src/main/python/resource_management/resources/service.py

@@ -0,0 +1,23 @@
+__all__ = ["Service"]
+
+from resource_management.base import Resource, ResourceArgument, BooleanArgument
+
+
+class Service(Resource):
+  service_name = ResourceArgument(default=lambda obj: obj.name)
+  enabled = ResourceArgument()
+  running = ResourceArgument()
+  pattern = ResourceArgument()
+  start_command = ResourceArgument()
+  stop_command = ResourceArgument()
+  restart_command = ResourceArgument()
+  reload_command = ResourceArgument()
+  status_command = ResourceArgument()
+  supports_restart = BooleanArgument(
+    default=lambda obj: bool(obj.restart_command))
+  supports_reload = BooleanArgument(
+    default=lambda obj: bool(obj.reload_command))
+  supports_status = BooleanArgument(
+    default=lambda obj: bool(obj.status_command))
+
+  actions = ["nothing", "start", "stop", "restart", "reload"]

ambari-agent/src/main/python/resource_management/resources/system.py

@@ -0,0 +1,74 @@
+__all__ = ["File", "Directory", "Link", "Execute", "Script", "Mount"]
+
+from resource_management.base import Resource, ForcedListArgument, ResourceArgument, BooleanArgument
+
+
+class File(Resource):
+  action = ForcedListArgument(default="create")
+  path = ResourceArgument(default=lambda obj: obj.name)
+  backup = ResourceArgument()
+  mode = ResourceArgument()
+  owner = ResourceArgument()
+  group = ResourceArgument()
+  content = ResourceArgument()
+
+  actions = Resource.actions + ["create", "delete", "touch"]
+
+
+class Directory(Resource):
+  action = ForcedListArgument(default="create")
+  path = ResourceArgument(default=lambda obj: obj.name)
+  mode = ResourceArgument()
+  owner = ResourceArgument()
+  group = ResourceArgument()
+  recursive = BooleanArgument(default=False)
+
+  actions = Resource.actions + ["create", "delete"]
+
+
+class Link(Resource):
+  action = ForcedListArgument(default="create")
+  path = ResourceArgument(default=lambda obj: obj.name)
+  to = ResourceArgument(required=True)
+  hard = BooleanArgument(default=False)
+
+  actions = Resource.actions + ["create", "delete"]
+
+
+class Execute(Resource):
+  action = ForcedListArgument(default="run")
+  command = ResourceArgument(default=lambda obj: obj.name)
+  creates = ResourceArgument()
+  cwd = ResourceArgument()
+  environment = ResourceArgument()
+  user = ResourceArgument()
+  group = ResourceArgument()
+  returns = ForcedListArgument(default=0)
+  timeout = ResourceArgument()
+
+  actions = Resource.actions + ["run"]
+
+
+class Script(Resource):
+  action = ForcedListArgument(default="run")
+  code = ResourceArgument(required=True)
+  cwd = ResourceArgument()
+  environment = ResourceArgument()
+  interpreter = ResourceArgument(default="/bin/bash")
+  user = ResourceArgument()
+  group = ResourceArgument()
+
+  actions = Resource.actions + ["run"]
+
+
+class Mount(Resource):
+  action = ForcedListArgument(default="mount")
+  mount_point = ResourceArgument(default=lambda obj: obj.name)
+  device = ResourceArgument()
+  fstype = ResourceArgument()
+  options = ResourceArgument(default=["defaults"])
+  dump = ResourceArgument(default=0)
+  passno = ResourceArgument(default=2)
+
+  actions = Resource.actions + ["mount", "umount", "remount", "enable",
+                                "disable"]

ambari-agent/src/main/python/resource_management/source.py

@@ -0,0 +1,120 @@
+from __future__ import with_statement
+from resource_management import environment
+
+__all__ = ["Source", "Template", "StaticFile", "DownloadSource"]
+
+import hashlib
+import os
+import urllib2
+import urlparse
+from resource_management.exceptions import Fail
+
+
+class Source(object):
+  def get_content(self):
+    raise NotImplementedError()
+
+  def get_checksum(self):
+    return None
+
+  def __call__(self):
+    return self.get_content()
+
+
+class StaticFile(Source):
+  def __init__(self, name, env=None):
+    self.name = name
+    self.env = env or environment.Environment.get_instance()
+
+  def get_content(self):
+    try:
+      cookbook, name = self.name.split('/', 1)
+    except ValueError:
+      raise Fail(
+        "[StaticFile(%s)] Path must include cookbook name (e.g. 'nginx/nginx.conf')" % self.name)
+    cb = self.env.cookbooks[cookbook]
+    path = os.path.join(cb.path, "files", name)
+    with open(path, "rb") as fp:
+      return fp.read()
+
+
+try:
+  from jinja2 import Environment, BaseLoader, TemplateNotFound
+except ImportError:
+  class Template(Source):
+    def __init__(self, name, variables=None, env=None):
+      raise Exception("Jinja2 required for Template")
+else:
+  class TemplateLoader(BaseLoader):
+    def __init__(self, env=None):
+      self.env = env or environment.Environment.get_instance()
+
+    def get_source(self, environment, template):
+      try:
+        cookbook, name = template.split('/', 1)
+      except ValueError:
+        raise Fail(
+          "[Template(%s)] Path must include cookbook name (e.g. 'nginx/nginx.conf.j2')" % template)
+      cb = self.env.cookbooks[cookbook]
+      path = os.path.join(cb.path, "templates", name)
+      if not os.path.exists(path):
+        raise TemplateNotFound("%s at %s" % (template, path))
+      mtime = os.path.getmtime(path)
+      with open(path, "rb") as fp:
+        source = fp.read().decode('utf-8')
+      return source, path, lambda: mtime == os.path.getmtime(path)
+
+  class Template(Source):
+    def __init__(self, name, variables=None, env=None):
+      self.name = name
+      self.env = env or environment.Environment.get_instance()
+      self.context = variables.copy() if variables else {}
+      self.template_env = Environment(loader=TemplateLoader(self.env),
+                                      autoescape=False)
+      self.template = self.template_env.get_template(self.name)
+
+    def get_content(self):
+      self.context.update(
+        env=self.env,
+        repr=repr,
+        str=str,
+        bool=bool,
+      )
+      rendered = self.template.render(self.context)
+      return rendered + "\n" if not rendered.endswith('\n') else rendered
+
+
+class DownloadSource(Source):
+  def __init__(self, url, cache=True, md5sum=None, env=None):
+    self.env = env or environment.Environment.get_instance()
+    self.url = url
+    self.md5sum = md5sum
+    self.cache = cache
+    if not 'download_path' in env.config:
+      env.config.download_path = '/var/tmp/downloads'
+    if not os.path.exists(env.config.download_path):
+      os.makedirs(self.env.config.download_path)
+
+  def get_content(self):
+    filepath = os.path.basename(urlparse.urlparse(self.url).path)
+    content = None
+    if not self.cache or not os.path.exists(
+      os.path.join(self.env.config.download_path, filepath)):
+      web_file = urllib2.urlopen(self.url)
+      content = web_file.read()
+    else:
+      update = False
+      with open(os.path.join(self.env.config.download_path, filepath)) as fp:
+        content = fp.read()
+      if self.md5sum:
+        m = hashlib.md5(content)
+        md5 = m.hexdigest()
+        if md5 != self.md5sum:
+          web_file = urllib2.urlopen(self.url)
+          content = web_file.read()
+          update = True
+      if self.cache and update:
+        with open(os.path.join(self.env.config.download_path, filepath),
+                  'w') as fp:
+          fp.write(content)
+    return content

ambari-agent/src/main/python/resource_management/system.py

@@ -0,0 +1,132 @@
+__all__ = ["System"]
+
+import os
+import sys
+from functools import wraps
+from subprocess import Popen, PIPE
+
+
+def lazy_property(undecorated):
+  name = '_' + undecorated.__name__
+
+  @property
+  @wraps(undecorated)
+  def decorated(self):
+    try:
+      return getattr(self, name)
+    except AttributeError:
+      v = undecorated(self)
+      setattr(self, name, v)
+      return v
+
+  return decorated
+
+
+class System(object):
+  @lazy_property
+  def os(self):
+    platform = sys.platform
+    if platform.startswith('linux'):
+      return "linux"
+    elif platform == "darwin":
+      return "darwin"
+    else:
+      return "unknown"
+
+  def unquote(self, val):
+    if val[0] == '"':
+      val = val[1:-1]
+    return val
+
+  @lazy_property
+  def arch(self):
+    machine = self.machine
+    if machine in ("i386", "i486", "i686"):
+      return "x86_32"
+    return machine
+
+  @lazy_property
+  def machine(self):
+    p = Popen(["/bin/uname", "-m"], stdout=PIPE, stderr=PIPE)
+    return p.communicate()[0].strip()
+
+  @lazy_property
+  def lsb(self):
+    if os.path.exists("/etc/lsb-release"):
+      with open("/etc/lsb-release", "rb") as fp:
+        lsb = (x.split('=') for x in fp.read().strip().split('\n'))
+      return dict(
+        (k.split('_', 1)[-1].lower(), self.unquote(v)) for k, v in lsb)
+    elif os.path.exists("/usr/bin/lsb_release"):
+      p = Popen(["/usr/bin/lsb_release", "-a"], stdout=PIPE, stderr=PIPE)
+      lsb = {}
+      for l in p.communicate()[0].split('\n'):
+        v = l.split(':', 1)
+        if len(v) != 2:
+          continue
+        lsb[v[0].strip().lower()] = self.unquote(v[1].strip().lower())
+      lsb['id'] = lsb.pop('distributor id')
+      return lsb
+
+  @lazy_property
+  def platform(self):
+    operatingsystem = self.os
+    if operatingsystem == "linux":
+      lsb = self.lsb
+      if not lsb:
+        if os.path.exists("/etc/redhat-release"):
+          return "redhat"
+        if os.path.exists("/etc/fedora-release"):
+          return "fedora"
+        if os.path.exists("/etc/debian_version"):
+          return "debian"
+        if os.path.exists("/etc/gentoo-release"):
+          return "gentoo"
+        if os.path.exists("/etc/system-release"):
+          with open("/etc/system-release", "rb") as fp:
+            release = fp.read()
+          if "Amazon Linux" in release:
+            return "amazon"
+        return "unknown"
+      return lsb['id'].lower()
+    elif operatingsystem == "darwin":
+      out = Popen("/usr/bin/sw_vers", stdout=PIPE).communicate()[0]
+      sw_vers = dict(
+        [y.strip() for y in x.split(':', 1)] for x in out.strip().split('\n'))
+      # ProductName, ProductVersion, BuildVersion
+      return sw_vers['ProductName'].lower().replace(' ', '_')
+    else:
+      return "unknown"
+
+  @lazy_property
+  def locales(self):
+    p = Popen("locale -a", shell=True, stdout=PIPE)
+    out = p.communicate()[0]
+    return out.strip().split("\n")
+
+  @lazy_property
+  def ec2(self):
+    if not os.path.exists("/proc/xen"):
+      return False
+    if os.path.exists("/etc/ec2_version"):
+      return True
+    return False
+
+  @lazy_property
+  def vm(self):
+    if os.path.exists("/usr/bin/VBoxControl"):
+      return "vbox"
+    elif os.path.exists("/usr/bin/vmware-toolbox-cmd") or os.path.exists(
+      "/usr/sbin/vmware-toolbox-cmd"):
+      return "vmware"
+    elif os.path.exists("/proc/xen"):
+      return "xen"
+    return None
+
+  @classmethod
+  def get_instance(cls):
+    try:
+      return cls._instance
+    except AttributeError:
+      cls._instance = cls()
+    return cls._instance

ambari-agent/src/main/python/resource_management/utils.py

@@ -0,0 +1,68 @@
+class AttributeDictionary(object):
+  def __init__(self, *args, **kwargs):
+    d = kwargs
+    if args:
+      d = args[0]
+    super(AttributeDictionary, self).__setattr__("_dict", d)
+
+  def __setattr__(self, name, value):
+    self[name] = value
+
+  def __getattr__(self, name):
+    if name in self.__dict__:
+      return self.__dict__[name]
+    try:
+      return self[name]
+    except KeyError:
+      raise AttributeError(
+        "'%s' object has no attribute '%s'" % (self.__class__.__name__, name))
+
+  def __setitem__(self, name, value):
+    self._dict[name] = self._convert_value(value)
+
+  def __getitem__(self, name):
+    return self._convert_value(self._dict[name])
+
+  def _convert_value(self, value):
+    if isinstance(value, dict) and not isinstance(value, AttributeDictionary):
+      return AttributeDictionary(value)
+    return value
+
+  def copy(self):
+    return self.__class__(self._dict.copy())
+
+  def update(self, *args, **kwargs):
+    self._dict.update(*args, **kwargs)
+
+  def items(self):
+    return self._dict.items()
+
+  def values(self):
+    return self._dict.values()
+
+  def keys(self):
+    return self._dict.keys()
+
+  def pop(self, *args, **kwargs):
+    return self._dict.pop(*args, **kwargs)
+
+  def get(self, *args, **kwargs):
+    return self._dict.get(*args, **kwargs)
+
+  def __repr__(self):
+    return self._dict.__repr__()
+
+  def __unicode__(self):
+    return self._dict.__unicode__()
+
+  def __str__(self):
+    return self._dict.__str__()
+
+  def __iter__(self):
+    return self._dict.__iter__()
+
+  def __getstate__(self):
+    return self._dict
+
+  def __setstate__(self, state):
+    super(AttributeDictionary, self).__setattr__("_dict", state)