You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ambari.apache.org by dm...@apache.org on 2013/11/01 16:28:19 UTC
[08/15] git commit: AMBARI-3658. Resource management. Support common
custom libraries. (Andrew Onischuk via dlysnichenko)
AMBARI-3658. Resource management. Support common custom libraries. (Andrew Onischuk via dlysnichenko)
Project: http://git-wip-us.apache.org/repos/asf/incubator-ambari/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ambari/commit/84274b4f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ambari/tree/84274b4f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ambari/diff/84274b4f
Branch: refs/heads/trunk
Commit: 84274b4f83d0c8ff0fbfb978ab95a1a7d79b29c7
Parents: 61d3ee9
Author: Lisnichenko Dmitro <dl...@hortonworks.com>
Authored: Fri Nov 1 17:15:25 2013 +0200
Committer: Lisnichenko Dmitro <dl...@hortonworks.com>
Committed: Fri Nov 1 17:15:25 2013 +0200
----------------------------------------------------------------------
.../main/python/resource_management/__init__.py | 10 +-
.../src/main/python/resource_management/base.py | 216 ----------------
.../python/resource_management/core/__init__.py | 9 +
.../python/resource_management/core/base.py | 216 ++++++++++++++++
.../resource_management/core/environment.py | 239 ++++++++++++++++++
.../resource_management/core/exceptions.py | 6 +
.../core/providers/__init__.py | 69 ++++++
.../core/providers/accounts.py | 97 ++++++++
.../resource_management/core/providers/mount.py | 117 +++++++++
.../core/providers/package/__init__.py | 36 +++
.../core/providers/package/yumrpm.py | 15 ++
.../core/providers/package/zypper.py | 15 ++
.../core/providers/service.py | 79 ++++++
.../core/providers/system.py | 248 +++++++++++++++++++
.../core/resources/__init__.py | 4 +
.../core/resources/accounts.py | 27 ++
.../core/resources/packaging.py | 12 +
.../core/resources/service.py | 16 ++
.../core/resources/system.py | 89 +++++++
.../python/resource_management/core/script.py | 102 ++++++++
.../python/resource_management/core/shell.py | 45 ++++
.../python/resource_management/core/source.py | 123 +++++++++
.../python/resource_management/core/system.py | 128 ++++++++++
.../python/resource_management/core/utils.py | 133 ++++++++++
.../python/resource_management/environment.py | 239 ------------------
.../python/resource_management/exceptions.py | 6 -
.../resource_management/libraries/__init__.py | 2 +
.../libraries/providers/__init__.py | 14 ++
.../libraries/resources/__init__.py | 0
.../resource_management/providers/__init__.py | 68 -----
.../resource_management/providers/accounts.py | 97 --------
.../resource_management/providers/mount.py | 117 ---------
.../providers/package/__init__.py | 36 ---
.../providers/package/yumrpm.py | 15 --
.../providers/package/zypper.py | 15 --
.../resource_management/providers/service.py | 79 ------
.../resource_management/providers/system.py | 248 -------------------
.../resource_management/resources/__init__.py | 4 -
.../resource_management/resources/accounts.py | 27 --
.../resource_management/resources/packaging.py | 12 -
.../resource_management/resources/service.py | 16 --
.../resource_management/resources/system.py | 89 -------
.../main/python/resource_management/script.py | 102 --------
.../main/python/resource_management/shell.py | 45 ----
.../main/python/resource_management/source.py | 124 ----------
.../main/python/resource_management/system.py | 128 ----------
.../main/python/resource_management/utils.py | 133 ----------
47 files changed, 1843 insertions(+), 1824 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ambari/blob/84274b4f/ambari-agent/src/main/python/resource_management/__init__.py
----------------------------------------------------------------------
diff --git a/ambari-agent/src/main/python/resource_management/__init__.py b/ambari-agent/src/main/python/resource_management/__init__.py
index 11d41de..afcd019 100644
--- a/ambari-agent/src/main/python/resource_management/__init__.py
+++ b/ambari-agent/src/main/python/resource_management/__init__.py
@@ -1,9 +1,3 @@
-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 *
+from resource_management.libraries import *
+from resource_management.core import *
-__version__ = "0.4.1"
http://git-wip-us.apache.org/repos/asf/incubator-ambari/blob/84274b4f/ambari-agent/src/main/python/resource_management/base.py
----------------------------------------------------------------------
diff --git a/ambari-agent/src/main/python/resource_management/base.py b/ambari-agent/src/main/python/resource_management/base.py
deleted file mode 100644
index 35aeefe..0000000
--- a/ambari-agent/src/main/python/resource_management/base.py
+++ /dev/null
@@ -1,216 +0,0 @@
-#!/usr/bin/env python
-
-__all__ = ["Resource", "ResourceArgument", "ForcedListArgument",
- "BooleanArgument"]
-
-import logging
-from resource_management.exceptions import Fail, InvalidArgument
-from resource_management.environment import Environment, Substitutor
-
-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
-
- log = logging.getLogger("resource_management.resource")
- is_updated = False
-
- action = ForcedListArgument(default="nothing")
- ignore_failures = BooleanArgument(default=False)
- notifies = ResourceArgument(default=[]) # this is not supported/recommended
- subscribes = ResourceArgument(default=[]) # this is not supported/recommended
- not_if = ResourceArgument() # pass command e.g. not_if = ('ls','/root/jdk')
- only_if = ResourceArgument() # pass command
- initial_wait = ResourceArgument() # in seconds
-
- actions = ["nothing"]
-
- def __new__(cls, name, env=None, provider=None, **kwargs):
- if isinstance(name, list):
- while len(name) != 1:
- cls(name.pop(0), env, provider, **kwargs)
-
- name = name[0]
-
- env = env or Environment.get_instance()
- name = Substitutor.substitute(name)
- 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 isinstance(name, list):
- name = name.pop(0)
-
- if hasattr(self, 'name'):
- return
-
- self.env = env or Environment.get_instance()
- self.name = Substitutor.substitute(name)
-
- self.provider = provider or getattr(self, 'provider', None)
-
- 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] = Substitutor.substitute(arg.validate(value))
- except InvalidArgument, exc:
- raise InvalidArgument("%s %s" % (self, exc))
-
- Resource.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']
-
- Resource.log = logging.getLogger("resource_management.resource")
-
- self.validate()
http://git-wip-us.apache.org/repos/asf/incubator-ambari/blob/84274b4f/ambari-agent/src/main/python/resource_management/core/__init__.py
----------------------------------------------------------------------
diff --git a/ambari-agent/src/main/python/resource_management/core/__init__.py b/ambari-agent/src/main/python/resource_management/core/__init__.py
new file mode 100644
index 0000000..cafe6f6
--- /dev/null
+++ b/ambari-agent/src/main/python/resource_management/core/__init__.py
@@ -0,0 +1,9 @@
+from resource_management.core.base import *
+from resource_management.core.environment import *
+from resource_management.core.exceptions import *
+from resource_management.core.providers import *
+from resource_management.core.resources import *
+from resource_management.core.source import *
+from resource_management.core.system import *
+
+__version__ = "0.4.1"
http://git-wip-us.apache.org/repos/asf/incubator-ambari/blob/84274b4f/ambari-agent/src/main/python/resource_management/core/base.py
----------------------------------------------------------------------
diff --git a/ambari-agent/src/main/python/resource_management/core/base.py b/ambari-agent/src/main/python/resource_management/core/base.py
new file mode 100644
index 0000000..fb2e26d
--- /dev/null
+++ b/ambari-agent/src/main/python/resource_management/core/base.py
@@ -0,0 +1,216 @@
+#!/usr/bin/env python
+
+__all__ = ["Resource", "ResourceArgument", "ForcedListArgument",
+ "BooleanArgument"]
+
+import logging
+from resource_management.core.exceptions import Fail, InvalidArgument
+from resource_management.core.environment import Environment, Substitutor
+
+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
+
+ log = logging.getLogger("resource_management.resource")
+ is_updated = False
+
+ action = ForcedListArgument(default="nothing")
+ ignore_failures = BooleanArgument(default=False)
+ notifies = ResourceArgument(default=[]) # this is not supported/recommended
+ subscribes = ResourceArgument(default=[]) # this is not supported/recommended
+ not_if = ResourceArgument() # pass command e.g. not_if = ('ls','/root/jdk')
+ only_if = ResourceArgument() # pass command
+ initial_wait = ResourceArgument() # in seconds
+
+ actions = ["nothing"]
+
+ def __new__(cls, name, env=None, provider=None, **kwargs):
+ if isinstance(name, list):
+ while len(name) != 1:
+ cls(name.pop(0), env, provider, **kwargs)
+
+ name = name[0]
+
+ env = env or Environment.get_instance()
+ name = Substitutor.substitute(name)
+ 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 isinstance(name, list):
+ name = name.pop(0)
+
+ if hasattr(self, 'name'):
+ return
+
+ self.env = env or Environment.get_instance()
+ self.name = Substitutor.substitute(name)
+
+ self.provider = provider or getattr(self, 'provider', None)
+
+ 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] = Substitutor.substitute(arg.validate(value))
+ except InvalidArgument, exc:
+ raise InvalidArgument("%s %s" % (self, exc))
+
+ Resource.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']
+
+ Resource.log = logging.getLogger("resource_management.resource")
+
+ self.validate()
http://git-wip-us.apache.org/repos/asf/incubator-ambari/blob/84274b4f/ambari-agent/src/main/python/resource_management/core/environment.py
----------------------------------------------------------------------
diff --git a/ambari-agent/src/main/python/resource_management/core/environment.py b/ambari-agent/src/main/python/resource_management/core/environment.py
new file mode 100644
index 0000000..8746ce8
--- /dev/null
+++ b/ambari-agent/src/main/python/resource_management/core/environment.py
@@ -0,0 +1,239 @@
+#!/usr/bin/env python
+
+__all__ = ["Environment"]
+
+import logging
+import os
+import shutil
+import time
+from datetime import datetime
+
+from resource_management.core import shell
+from resource_management.core.exceptions import Fail
+from resource_management.core.providers import find_provider
+from resource_management.core.utils import AttributeDictionary, ParamsAttributeDictionary
+from resource_management.core.system import System
+from string import Template
+
+
+class Environment(object):
+ _instances = []
+
+ def __init__(self, basedir=None, params=None):
+ """
+ @param basedir: basedir/files, basedir/templates are the places where templates / static files
+ are looked up
+ @param params: configurations dictionary (this will be accessible in the templates)
+ """
+ self.log = logging.getLogger("resource_management")
+ self.reset(basedir, params)
+
+ def reset(self, basedir, params):
+ self.system = System.get_instance()
+ self.config = AttributeDictionary()
+ self.resources = {}
+ self.resource_list = []
+ Substitutor.default_prefixes = []
+ self.delayed_actions = set()
+ self.update_config({
+ # current time
+ 'date': datetime.now(),
+ # backups here files which were rewritten while executing File resource
+ 'backup.path': '/tmp/resource_management/backup',
+ # prefix for this files
+ 'backup.prefix': datetime.now().strftime("%Y%m%d%H%M%S"),
+ # dir where templates,failes dirs are
+ 'basedir': basedir,
+ # variables, which can be used in templates
+ 'params': ParamsAttributeDictionary(Substitutor, params),
+ })
+
+ def backup_file(self, path):
+ if self.config.backup:
+ if not os.path.exists(self.config.backup.path):
+ os.makedirs(self.config.backup.path, 0700)
+ new_name = self.config.backup.prefix + path.replace('/', '-')
+ backup_path = os.path.join(self.config.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 set_default_prefixes(self, dict):
+ Substitutor.default_prefixes = dict
+
+ def _check_condition(self, cond):
+ if hasattr(cond, '__call__'):
+ return cond()
+
+ if isinstance(cond, basestring):
+ ret, out = shell.call(cond)
+ 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.initial_wait:
+ time.sleep(resource.initial_wait)
+
+ 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:
+ if not resource.ignore_failures:
+ self.run_action(resource, action)
+ else:
+ try:
+ self.run_action(resource, action)
+ except Exception:
+ pass
+
+ # 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']
+
+
+class Substitutor():
+ log = logging.getLogger("resource_management.resource")
+ default_prefixes = []
+
+ class ExtendedTemplate(Template):
+ """
+ This is done to support substitution of dictionaries in dictionaries
+ ( ':' sign)
+
+ default is:
+ idpattern = r'[_a-z][_a-z0-9]*'
+ """
+ idpattern = r'[_a-z][_a-z0-9:]*'
+
+ @staticmethod
+ def _get_subdict(name, dic):
+ """
+ "a:b:c" => a[b][c]
+
+ doesn't use prefixes
+ """
+ name_parts = name.split(':')
+ curr = dic
+
+ for x in name_parts:
+ curr = curr[x]
+ return curr
+
+ @staticmethod
+ def get_subdict(name, dic):
+ """
+ "a:b:c" => a[b][c]
+
+ can use prefixes
+ """
+ prefixes = list(Substitutor.default_prefixes)
+ prefixes.insert(0, None) # for not prefixed case
+ name_parts = name.split(':')
+ is_found = False
+ result = None
+
+ for prefix in prefixes:
+ curr = Substitutor._get_subdict(prefix,dic) if prefix else dic
+
+ try:
+ for x in name_parts:
+ curr = curr[x]
+ except (KeyError, TypeError):
+ continue
+
+ if is_found:
+ raise Fail("Variable ${%s} found more than one time, please check your default prefixes!" % name)
+
+ is_found = True
+ result = curr
+
+ if not result:
+ raise Fail("Configuration on ${%s} cannot be resolved" % name)
+
+ return result
+
+ @staticmethod
+ def substitute(val):
+ env = Environment.get_instance()
+ dic = env.config.params
+
+ if dic and isinstance(val, str):
+ result = Substitutor.ExtendedTemplate(val).substitute(dic)
+ if '$' in val:
+ Substitutor.log.debug("%s after substitution is %s", val, result)
+ return result
+
+ return val
http://git-wip-us.apache.org/repos/asf/incubator-ambari/blob/84274b4f/ambari-agent/src/main/python/resource_management/core/exceptions.py
----------------------------------------------------------------------
diff --git a/ambari-agent/src/main/python/resource_management/core/exceptions.py b/ambari-agent/src/main/python/resource_management/core/exceptions.py
new file mode 100644
index 0000000..26cd3a2
--- /dev/null
+++ b/ambari-agent/src/main/python/resource_management/core/exceptions.py
@@ -0,0 +1,6 @@
+class Fail(Exception):
+ pass
+
+
+class InvalidArgument(Fail):
+ pass
http://git-wip-us.apache.org/repos/asf/incubator-ambari/blob/84274b4f/ambari-agent/src/main/python/resource_management/core/providers/__init__.py
----------------------------------------------------------------------
diff --git a/ambari-agent/src/main/python/resource_management/core/providers/__init__.py b/ambari-agent/src/main/python/resource_management/core/providers/__init__.py
new file mode 100644
index 0000000..7cf76d0
--- /dev/null
+++ b/ambari-agent/src/main/python/resource_management/core/providers/__init__.py
@@ -0,0 +1,69 @@
+__all__ = ["Provider", "find_provider"]
+
+import logging
+from resource_management.core.exceptions import Fail
+from resource_management.libraries.providers import PROVIDERS as LIBRARY_PROVIDERS
+
+
+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(
+ redhat=dict(
+ Package="resource_management.core.providers.package.yumrpm.YumProvider",
+ ),
+ centos=dict(
+ Package="resource_management.core.providers.package.yumrpm.YumProvider",
+ ),
+ suse=dict(
+ Package="resource_management.core.providers.package.zypper.ZypperProvider",
+ ),
+ fedora=dict(
+ Package="resource_management.core.providers.package.yumrpm.YumProvider",
+ ),
+ amazon=dict(
+ Package="resource_management.core.providers.package.yumrpm.YumProvider",
+ ),
+ default=dict(
+ File="resource_management.core.providers.system.FileProvider",
+ Directory="resource_management.core.providers.system.DirectoryProvider",
+ Link="resource_management.core.providers.system.LinkProvider",
+ Execute="resource_management.core.providers.system.ExecuteProvider",
+ Script="resource_management.core.providers.system.ScriptProvider",
+ Mount="resource_management.core.providers.mount.MountProvider",
+ User="resource_management.core.providers.accounts.UserProvider",
+ Group="resource_management.core.providers.accounts.GroupProvider",
+ Service="resource_management.core.providers.service.ServiceProvider",
+ ),
+)
+
+
+def find_provider(env, resource, class_path=None):
+ if not class_path:
+ providers = [PROVIDERS, LIBRARY_PROVIDERS]
+ for provider in providers:
+ if resource in provider[env.system.platform]:
+ class_path = provider[env.system.platform][resource]
+ break
+ if resource in provider["default"]:
+ class_path = provider["default"][resource]
+ break
+
+ 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)
http://git-wip-us.apache.org/repos/asf/incubator-ambari/blob/84274b4f/ambari-agent/src/main/python/resource_management/core/providers/accounts.py
----------------------------------------------------------------------
diff --git a/ambari-agent/src/main/python/resource_management/core/providers/accounts.py b/ambari-agent/src/main/python/resource_management/core/providers/accounts.py
new file mode 100644
index 0000000..583e6c3
--- /dev/null
+++ b/ambari-agent/src/main/python/resource_management/core/providers/accounts.py
@@ -0,0 +1,97 @@
+from __future__ import with_statement
+
+import grp
+import pwd
+from resource_management.core import shell
+from resource_management.core.providers import Provider
+
+
+class UserProvider(Provider):
+ def action_create(self):
+ if not self.user:
+ command = ['useradd', "-m"]
+ self.log.info("Adding user %s" % self.resource)
+ else:
+ command = ['usermod']
+ self.log.info("Modifying user %s" % (self.resource.username))
+
+ options = dict(
+ comment="-c",
+ gid="-g",
+ uid="-u",
+ shell="-s",
+ password="-p",
+ home="-d",
+ )
+
+ if self.resource.system and not self.user:
+ command.append("--system")
+
+ if self.resource.groups:
+ command += ["-G", ",".join(self.resource.groups)]
+
+ for option_name, option_flag in 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)
+
+ shell.checked_call(command)
+ self.resource.updated()
+
+ def action_remove(self):
+ if self.user:
+ command = ['userdel', self.resource.username]
+ shell.checked_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']
+ self.log.info("Adding group %s" % self.resource)
+ else:
+ command = ['groupmod']
+ self.log.info("Modifying group %s" % (self.resource.group_name))
+
+ options = dict(
+ gid="-g",
+ password="-p",
+ )
+
+ for option_name, option_flag in 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)
+
+ shell.checked_call(command)
+ self.resource.updated()
+
+ group = self.group
+
+ def action_remove(self):
+ if self.group:
+ command = ['groupdel', self.resource.group_name]
+ shell.checked_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
http://git-wip-us.apache.org/repos/asf/incubator-ambari/blob/84274b4f/ambari-agent/src/main/python/resource_management/core/providers/mount.py
----------------------------------------------------------------------
diff --git a/ambari-agent/src/main/python/resource_management/core/providers/mount.py b/ambari-agent/src/main/python/resource_management/core/providers/mount.py
new file mode 100644
index 0000000..446bba0
--- /dev/null
+++ b/ambari-agent/src/main/python/resource_management/core/providers/mount.py
@@ -0,0 +1,117 @@
+from __future__ import with_statement
+
+import os
+import re
+from resource_management.core.base import Fail
+from resource_management.core.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
http://git-wip-us.apache.org/repos/asf/incubator-ambari/blob/84274b4f/ambari-agent/src/main/python/resource_management/core/providers/package/__init__.py
----------------------------------------------------------------------
diff --git a/ambari-agent/src/main/python/resource_management/core/providers/package/__init__.py b/ambari-agent/src/main/python/resource_management/core/providers/package/__init__.py
new file mode 100644
index 0000000..e862cd1
--- /dev/null
+++ b/ambari-agent/src/main/python/resource_management/core/providers/package/__init__.py
@@ -0,0 +1,36 @@
+from resource_management.core.base import Fail
+from resource_management.core.providers import Provider
+
+
+class PackageProvider(Provider):
+ def __init__(self, *args, **kwargs):
+ super(PackageProvider, self).__init__(*args, **kwargs)
+
+ def install_package(self, name, version):
+ raise NotImplementedError()
+ def remove_package(self, name):
+ raise NotImplementedError()
+ def upgrade_package(self, name, version):
+ raise NotImplementedError()
+
+ def action_install(self):
+ package_name = self.get_package_name_with_version()
+ self.log.info("Installing package %s", package_name)
+ self.install_package(package_name)
+
+ def action_upgrade(self):
+ package_name = self.get_package_name_with_version()
+ self.log.info("Upgrading package %s", package_name)
+ self.upgrade_package(package_name)
+
+ def action_remove(self):
+ package_name = self.get_package_name_with_version()
+ self.log.info("Removing package %s", package_name)
+ self.remove_package(package_name)
+
+ def get_package_name_with_version(self):
+ if self.resource.version:
+ return self.resource.package_name + '-' + self.resource.version
+ else:
+ return self.resource.package_name
+
http://git-wip-us.apache.org/repos/asf/incubator-ambari/blob/84274b4f/ambari-agent/src/main/python/resource_management/core/providers/package/yumrpm.py
----------------------------------------------------------------------
diff --git a/ambari-agent/src/main/python/resource_management/core/providers/package/yumrpm.py b/ambari-agent/src/main/python/resource_management/core/providers/package/yumrpm.py
new file mode 100644
index 0000000..0758fb4
--- /dev/null
+++ b/ambari-agent/src/main/python/resource_management/core/providers/package/yumrpm.py
@@ -0,0 +1,15 @@
+from resource_management.core.providers.package import PackageProvider
+from resource_management.core import shell
+
+INSTALL_CMD = "/usr/bin/yum -d 0 -e 0 -y install %s"
+REMOVE_CMD = "/usr/bin/yum -d 0 -e 0 -y erase %s"
+
+class YumProvider(PackageProvider):
+ def install_package(self, name):
+ shell.checked_call(INSTALL_CMD % (name))
+
+ def upgrade_package(self, name):
+ return self.install_package(name)
+
+ def remove_package(self, name):
+ shell.checked_call(REMOVE_CMD % (name))
http://git-wip-us.apache.org/repos/asf/incubator-ambari/blob/84274b4f/ambari-agent/src/main/python/resource_management/core/providers/package/zypper.py
----------------------------------------------------------------------
diff --git a/ambari-agent/src/main/python/resource_management/core/providers/package/zypper.py b/ambari-agent/src/main/python/resource_management/core/providers/package/zypper.py
new file mode 100644
index 0000000..0099a65
--- /dev/null
+++ b/ambari-agent/src/main/python/resource_management/core/providers/package/zypper.py
@@ -0,0 +1,15 @@
+from resource_management.core.providers.package import PackageProvider
+from resource_management.core import shell
+
+INSTALL_CMD = "/usr/bin/zypper --quiet install --auto-agree-with-licenses --no-confirm %s"
+REMOVE_CMD = "/usr/bin/zypper --quiet remove --no-confirm %s"
+
+class ZypperProvider(PackageProvider):
+ def install_package(self, name):
+ shell.checked_call(INSTALL_CMD % (name))
+
+ def upgrade_package(self, name):
+ return self.install_package(name)
+
+ def remove_package(self, name):
+ shell.checked_call(REMOVE_CMD % (name))
http://git-wip-us.apache.org/repos/asf/incubator-ambari/blob/84274b4f/ambari-agent/src/main/python/resource_management/core/providers/service.py
----------------------------------------------------------------------
diff --git a/ambari-agent/src/main/python/resource_management/core/providers/service.py b/ambari-agent/src/main/python/resource_management/core/providers/service.py
new file mode 100644
index 0000000..17740f9
--- /dev/null
+++ b/ambari-agent/src/main/python/resource_management/core/providers/service.py
@@ -0,0 +1,79 @@
+import os
+
+from resource_management.core import shell
+from resource_management.core.base import Fail
+from resource_management.core.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,out = shell.call(custom_cmd)
+ else:
+ ret = self._init_cmd(command)
+
+ if expect is not None and expect != ret:
+ raise Fail("%r command %s for service %s failed with return code: %d. %s" % (
+ self, command, self.resource.service_name, ret, out))
+ return ret
+
+ def _init_cmd(self, command):
+ if self._upstart:
+ if command == "status":
+ ret,out = shell.call(["/sbin/" + command, self.resource.service_name])
+ _proc, state = out.strip().split(' ', 1)
+ ret = 0 if state != "stop/waiting" else 1
+ else:
+ ret,out = shell.call(["/sbin/" + command, self.resource.service_name])
+ else:
+ ret,out = shell.call(["/etc/init.d/%s" % self.resource.service_name, command])
+ 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
http://git-wip-us.apache.org/repos/asf/incubator-ambari/blob/84274b4f/ambari-agent/src/main/python/resource_management/core/providers/system.py
----------------------------------------------------------------------
diff --git a/ambari-agent/src/main/python/resource_management/core/providers/system.py b/ambari-agent/src/main/python/resource_management/core/providers/system.py
new file mode 100644
index 0000000..18f624c
--- /dev/null
+++ b/ambari-agent/src/main/python/resource_management/core/providers/system.py
@@ -0,0 +1,248 @@
+from __future__ import with_statement
+
+import grp
+import os
+import pwd
+import time
+import shutil
+from resource_management.core import shell
+from resource_management.core.base import Fail
+from resource_management.core.providers import Provider
+
+
+def _coerce_uid(user):
+ try:
+ uid = int(user)
+ except ValueError:
+ try:
+ uid = pwd.getpwnam(user).pw_uid
+ except KeyError:
+ raise Fail("User %s doesn't exist." % user)
+ return uid
+
+
+def _coerce_gid(group):
+ try:
+ gid = int(group)
+ except ValueError:
+ try:
+ gid = grp.getgrnam(group).gr_gid
+ except KeyError:
+ raise Fail("Group %s doesn't exist." % group)
+ 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
+
+ if os.path.isdir(path):
+ raise Fail("Applying %s failed, directory with name %s exists" % (self.resource, path))
+
+ dirname = os.path.dirname(path)
+ if not os.path.isdir(dirname):
+ raise Fail("Applying %s failed, parent directory %s doesn't exist" % (self.resource, dirname))
+
+ write = False
+ content = self._get_content()
+ if not os.path.exists(path):
+ write = True
+ reason = "it doesn't exist"
+ elif self.resource.replace:
+ 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"
+ if self.resource.backup:
+ 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.isdir(path):
+ raise Fail("Applying %s failed, %s is directory not file!" % (self.resource, path))
+
+ if os.path.exists(path):
+ self.log.info("Deleting %s" % self.resource)
+ os.unlink(path)
+ self.resource.updated()
+
+ 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 not os.path.isdir(path):
+ raise Fail("Applying %s failed, file %s already exists" % (self.resource, path))
+
+ 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):
+ if not os.path.isdir(path):
+ raise Fail("Applying %s failed, %s is not a directory" % (self.resource, path))
+
+ self.log.info("Removing directory %s and all its content" % self.resource)
+ shutil.rmtree(path)
+ 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.resource, oldpath))
+ os.unlink(path)
+
+ if self.resource.hard:
+ if not os.path.exists(self.resource.to):
+ raise Fail("Failed to apply %s, linking to nonexistent location %s" % (self.resource, self.resource.to))
+ if os.path.isdir(self.resource.to):
+ raise Fail("Failed to apply %s, cannot create hard link to a directory (%s)" % (self.resource, self.resource.to))
+
+ self.log.info("Creating hard %s" % self.resource)
+ os.link(self.resource.to, path)
+ self.resource.updated()
+ else:
+ if not os.path.exists(self.resource.to):
+ self.log.info("Warning: linking to nonexistent location %s", self.resource.to)
+
+ 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)
+
+ if self.resource.path != []:
+ self.resource.environment['PATH'] = os.pathsep.join(self.resource.path)
+
+ for i in range (0, self.resource.tries):
+ try:
+ shell.checked_call(self.resource.command, logoutput=self.resource.logoutput,
+ cwd=self.resource.cwd, env=self.resource.environment,
+ preexec_fn=_preexec_fn(self.resource))
+ break
+ except Fail as ex:
+ if i == self.resource.tries-1: # last try
+ raise ex
+ else:
+ self.log.info("Retrying after %d seconds. Reason: %s", self.resource.try_sleep, str(ex))
+ time.sleep(self.resource.try_sleep)
+
+ 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)
+ shell.call([self.resource.interpreter, tf.name],
+ cwd=self.resource.cwd, env=self.resource.environment,
+ preexec_fn=_preexec_fn(self.resource))
+ self.resource.updated()
http://git-wip-us.apache.org/repos/asf/incubator-ambari/blob/84274b4f/ambari-agent/src/main/python/resource_management/core/resources/__init__.py
----------------------------------------------------------------------
diff --git a/ambari-agent/src/main/python/resource_management/core/resources/__init__.py b/ambari-agent/src/main/python/resource_management/core/resources/__init__.py
new file mode 100644
index 0000000..c4150ed
--- /dev/null
+++ b/ambari-agent/src/main/python/resource_management/core/resources/__init__.py
@@ -0,0 +1,4 @@
+from resource_management.core.resources.accounts import *
+from resource_management.core.resources.packaging import *
+from resource_management.core.resources.service import *
+from resource_management.core.resources.system import *
http://git-wip-us.apache.org/repos/asf/incubator-ambari/blob/84274b4f/ambari-agent/src/main/python/resource_management/core/resources/accounts.py
----------------------------------------------------------------------
diff --git a/ambari-agent/src/main/python/resource_management/core/resources/accounts.py b/ambari-agent/src/main/python/resource_management/core/resources/accounts.py
new file mode 100644
index 0000000..cd86de2
--- /dev/null
+++ b/ambari-agent/src/main/python/resource_management/core/resources/accounts.py
@@ -0,0 +1,27 @@
+__all__ = ["Group", "User"]
+
+from resource_management.core.base import Resource, ForcedListArgument, ResourceArgument, BooleanArgument
+
+
+class Group(Resource):
+ action = ForcedListArgument(default="create")
+ group_name = ResourceArgument(default=lambda obj: obj.name)
+ gid = ResourceArgument()
+ password = ResourceArgument()
+
+ actions = Resource.actions + ["create", "remove"]
+
+
+class User(Resource):
+ action = ForcedListArgument(default="create")
+ username = ResourceArgument(default=lambda obj: obj.name)
+ comment = ResourceArgument()
+ uid = ResourceArgument()
+ gid = ResourceArgument()
+ groups = ForcedListArgument(default=[]) # supplementary groups
+ home = ResourceArgument()
+ shell = ResourceArgument(default="/bin/bash")
+ password = ResourceArgument()
+ system = BooleanArgument(default=False)
+
+ actions = Resource.actions + ["create", "remove"]
http://git-wip-us.apache.org/repos/asf/incubator-ambari/blob/84274b4f/ambari-agent/src/main/python/resource_management/core/resources/packaging.py
----------------------------------------------------------------------
diff --git a/ambari-agent/src/main/python/resource_management/core/resources/packaging.py b/ambari-agent/src/main/python/resource_management/core/resources/packaging.py
new file mode 100644
index 0000000..8ca4534
--- /dev/null
+++ b/ambari-agent/src/main/python/resource_management/core/resources/packaging.py
@@ -0,0 +1,12 @@
+__all__ = ["Package"]
+
+from resource_management.core.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"]
+ build_vars = ForcedListArgument(default=[])
http://git-wip-us.apache.org/repos/asf/incubator-ambari/blob/84274b4f/ambari-agent/src/main/python/resource_management/core/resources/service.py
----------------------------------------------------------------------
diff --git a/ambari-agent/src/main/python/resource_management/core/resources/service.py b/ambari-agent/src/main/python/resource_management/core/resources/service.py
new file mode 100644
index 0000000..94f03b2
--- /dev/null
+++ b/ambari-agent/src/main/python/resource_management/core/resources/service.py
@@ -0,0 +1,16 @@
+__all__ = ["Service"]
+
+from resource_management.core.base import Resource, ResourceArgument, ForcedListArgument
+
+
+class Service(Resource):
+ action = ForcedListArgument(default="start")
+ service_name = ResourceArgument(default=lambda obj: obj.name)
+ #enabled = ResourceArgument() # Maybe add support to put in/out autostart.
+ start_command = ResourceArgument()
+ stop_command = ResourceArgument()
+ restart_command = ResourceArgument()
+ reload_command = ResourceArgument() # reload the config file without interrupting pending operations
+ status_command = ResourceArgument()
+
+ actions = ["nothing", "start", "stop", "restart", "reload"]
http://git-wip-us.apache.org/repos/asf/incubator-ambari/blob/84274b4f/ambari-agent/src/main/python/resource_management/core/resources/system.py
----------------------------------------------------------------------
diff --git a/ambari-agent/src/main/python/resource_management/core/resources/system.py b/ambari-agent/src/main/python/resource_management/core/resources/system.py
new file mode 100644
index 0000000..4bcb0f5
--- /dev/null
+++ b/ambari-agent/src/main/python/resource_management/core/resources/system.py
@@ -0,0 +1,89 @@
+__all__ = ["File", "Directory", "Link", "Execute", "Script", "Mount"]
+
+from resource_management.core.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()
+ # whether to replace files with different content
+ replace = ResourceArgument(default=True)
+
+ actions = Resource.actions + ["create", "delete"]
+
+
+class Directory(Resource):
+ action = ForcedListArgument(default="create")
+ path = ResourceArgument(default=lambda obj: obj.name)
+ mode = ResourceArgument()
+ owner = ResourceArgument()
+ group = ResourceArgument()
+ recursive = BooleanArgument(default=False) # this work for 'create', 'delete' is anyway recursive
+
+ 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")
+
+ """
+ Recommended:
+ command = ('rm','-f','myfile')
+ Not recommended:
+ command = 'rm -f myfile'
+
+ The first one helps to stop escaping issues
+ """
+ command = ResourceArgument(default=lambda obj: obj.name)
+
+ creates = ResourceArgument()
+ cwd = ResourceArgument()
+ # this runs command with a specific env variables, env={'JAVA_HOME': '/usr/jdk'}
+ environment = ResourceArgument(default={})
+ user = ResourceArgument()
+ group = ResourceArgument()
+ returns = ForcedListArgument(default=0)
+ tries = ResourceArgument(default=1)
+ try_sleep = ResourceArgument(default=0) # seconds
+ path = ForcedListArgument(default=[])
+ actions = Resource.actions + ["run"]
+ logoutput = BooleanArgument(default=False)
+
+
+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"]
http://git-wip-us.apache.org/repos/asf/incubator-ambari/blob/84274b4f/ambari-agent/src/main/python/resource_management/core/script.py
----------------------------------------------------------------------
diff --git a/ambari-agent/src/main/python/resource_management/core/script.py b/ambari-agent/src/main/python/resource_management/core/script.py
new file mode 100644
index 0000000..d64010a
--- /dev/null
+++ b/ambari-agent/src/main/python/resource_management/core/script.py
@@ -0,0 +1,102 @@
+#!/usr/bin/env python2.6
+
+'''
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+'''
+
+import sys
+import json
+import logging
+
+from resource_management.environment import Environment
+from resource_management.exceptions import Fail
+
+
+class Script():
+ """
+ Executes a command for custom service. stdout and stderr are written to
+ tmpoutfile and to tmperrfile respectively.
+ """
+
+ def __init__(self):
+ pass
+
+
+ def start(self, env, params): # TODO: just for test runs; remove
+ env.set_prefixes("ddd")
+ print "Start!"
+ pass
+
+
+ def execute(self):
+ """
+ Sets up logging;
+ Parses command parameters and executes method relevant to command type
+ """
+ # set up logging (two separate loggers for stderr and stdout with different loglevels)
+ logger = logging.getLogger('resource_management')
+ logger.setLevel(logging.DEBUG)
+ formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+ chout = logging.StreamHandler(sys.stdout)
+ chout.setLevel(logging.DEBUG)
+ chout.setFormatter(formatter)
+ cherr = logging.StreamHandler(sys.stderr)
+ cherr.setLevel(logging.ERROR)
+ cherr.setFormatter(formatter)
+ logger.addHandler(cherr)
+ # parse arguments
+ if len(sys.argv) < 1+3:
+ logger.error("Script expects at least 3 arguments")
+ sys.exit(1)
+ command_type = str.lower(sys.argv[1])
+ # parse command parameters
+ command_data_file = sys.argv[2]
+ basedir = sys.argv[3]
+ try:
+ with open(command_data_file, "r") as f:
+ pass
+ params = json.load(f)
+ except IOError:
+ logger.exception("Can not read json file with command parameters: ")
+ sys.exit(1)
+ # Run class method mentioned by a command type
+ self_methods = dir(self)
+ if not command_type in self_methods:
+ logger.error("Script {0} has not method '{1}'".format(sys.argv[0], command_type))
+ sys.exit(1)
+ method = getattr(self, command_type)
+ try:
+ with Environment(basedir, params) as env:
+ method(env, params)
+ env.run()
+ except Fail:
+ logger.exception("Got exception while executing method '{0}':".format(command_type))
+ sys.exit(1)
+
+
+
+ def fail_with_error(self, message):
+ """
+ Prints error message and exits with non-zero exit code
+ """
+ print("Error: " + message)
+ sys.stderr.write("Error: " + message)
+ sys.exit(1)
+
+
+if __name__ == "__main__":
+ Script().execute()
http://git-wip-us.apache.org/repos/asf/incubator-ambari/blob/84274b4f/ambari-agent/src/main/python/resource_management/core/shell.py
----------------------------------------------------------------------
diff --git a/ambari-agent/src/main/python/resource_management/core/shell.py b/ambari-agent/src/main/python/resource_management/core/shell.py
new file mode 100644
index 0000000..0ebba09
--- /dev/null
+++ b/ambari-agent/src/main/python/resource_management/core/shell.py
@@ -0,0 +1,45 @@
+import logging
+import subprocess
+from exceptions import Fail
+
+log = logging.getLogger("resource_management.provider")
+
+def checked_call(command, logoutput=False,
+ cwd=None, env=None, preexec_fn=None):
+ return _call(command, logoutput, True, cwd, env, preexec_fn)
+
+def call(command, logoutput=False,
+ cwd=None, env=None, preexec_fn=None):
+ return _call(command, logoutput, False, cwd, env, preexec_fn)
+
+
+def _call(command, logoutput=False, throw_on_failure=True,
+ cwd=None, env=None, preexec_fn=None):
+ """
+ Execute shell command
+
+ @param command: list/tuple of arguments (recommended as more safe - don't need to escape)
+ or string of the command to execute
+ @param logoutput: boolean, whether command output should be logged of not
+ @param throw_on_failure: if true, when return code is not zero exception is thrown
+
+ @return: retrun_code, stdout
+ """
+
+ shell = not isinstance(command, (list, tuple))
+
+ proc = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
+ cwd=cwd, env=env, shell=shell,
+ preexec_fn=preexec_fn)
+
+ out = proc.communicate()[0]
+ code = proc.returncode
+
+ if logoutput and out and out!="":
+ log.info(out)
+
+ if throw_on_failure and code:
+ err_msg = ("Execution of '%s' returned %d. %s") % (command, code, out)
+ raise Fail(err_msg)
+
+ return code, out
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-ambari/blob/84274b4f/ambari-agent/src/main/python/resource_management/core/source.py
----------------------------------------------------------------------
diff --git a/ambari-agent/src/main/python/resource_management/core/source.py b/ambari-agent/src/main/python/resource_management/core/source.py
new file mode 100644
index 0000000..3a0035f
--- /dev/null
+++ b/ambari-agent/src/main/python/resource_management/core/source.py
@@ -0,0 +1,123 @@
+from __future__ import with_statement
+from resource_management.core import environment
+
+__all__ = ["Source", "Template", "StaticFile", "DownloadSource"]
+
+import hashlib
+import os
+import urllib2
+import urlparse
+
+
+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):
+ # absolute path
+ if self.name.startswith(os.path.sep):
+ path = self.name
+ # relative path
+ else:
+ basedir = self.env.config.basedir
+ path = os.path.join(basedir, "files", self.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_name):
+ # absolute path
+ if template_name.startswith(os.path.sep):
+ path = template_name
+ # relative path
+ else:
+ basedir = self.env.config.basedir
+ path = os.path.join(basedir, "templates", template_name)
+
+ if not os.path.exists(path):
+ raise TemplateNotFound("%s at %s" % (template_name, 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()
+ params = self.env.config.params
+ variables = params if params else variables
+ 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
http://git-wip-us.apache.org/repos/asf/incubator-ambari/blob/84274b4f/ambari-agent/src/main/python/resource_management/core/system.py
----------------------------------------------------------------------
diff --git a/ambari-agent/src/main/python/resource_management/core/system.py b/ambari-agent/src/main/python/resource_management/core/system.py
new file mode 100644
index 0000000..5239186
--- /dev/null
+++ b/ambari-agent/src/main/python/resource_management/core/system.py
@@ -0,0 +1,128 @@
+__all__ = ["System"]
+
+import os
+import sys
+from resource_management.core import shell
+from functools import wraps
+
+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):
+ code, out = shell.call(["/bin/uname", "-m"])
+ return out.strip()
+
+ @lazy_property
+ def lsb(self):
+ if os.path.exists("/usr/bin/lsb_release"):
+ code, out = shell.call(["/usr/bin/lsb_release", "-a"])
+ lsb = {}
+ for l in out.split('\n'):
+ v = l.split(':', 1)
+ if len(v) != 2:
+ continue
+ lsb[v[0].strip().lower()] = self.unquote(v[1].strip().lower())
+
+ # failsafe
+ if not 'distributor id' in lsb:
+ return None
+
+ lsb['id'] = lsb.pop('distributor id')
+ return lsb
+
+ return None
+
+ @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/centos-release"):
+ return "centos"
+ if os.path.exists("/etc/SuSE-release"):
+ return "suse"
+ 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"
+
+ lsb_id = lsb['id'].lower()
+ if lsb_id =="suse linux":
+ return "suse"
+ return lsb_id
+ return "unknown"
+
+ @lazy_property
+ def locales(self):
+ code, out = shell.call("locale -a")
+ 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
http://git-wip-us.apache.org/repos/asf/incubator-ambari/blob/84274b4f/ambari-agent/src/main/python/resource_management/core/utils.py
----------------------------------------------------------------------
diff --git a/ambari-agent/src/main/python/resource_management/core/utils.py b/ambari-agent/src/main/python/resource_management/core/utils.py
new file mode 100644
index 0000000..4a00576
--- /dev/null
+++ b/ambari-agent/src/main/python/resource_management/core/utils.py
@@ -0,0 +1,133 @@
+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)
+
+class ParamsAttributeDictionary(object):
+ """
+ This class can store user parameters
+ and support some features necessary for substitution to work.
+ """
+ def __init__(self, substitutor, *args, **kwargs):
+ d = kwargs
+ if len(args)==1:
+ d = args[0]
+ super(ParamsAttributeDictionary, self).__setattr__("_dict", d)
+ super(ParamsAttributeDictionary, self).__setattr__("substitutor", substitutor)
+
+ def __setattr__(self, name, value):
+ self[name] = value
+
+ def __setitem__(self, name, value):
+ self._dict[name] = self._convert_value(value)
+
+ def __getitem__(self, name):
+ val = self.substitutor.get_subdict(name, self._dict)
+ return self._convert_value(val)
+
+ def _convert_value(self, value):
+ if isinstance(value, dict) and not isinstance(value, ParamsAttributeDictionary):
+ return ParamsAttributeDictionary(self.substitutor, 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(ParamsAttributeDictionary, self).__setattr__("_dict", state)
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-ambari/blob/84274b4f/ambari-agent/src/main/python/resource_management/environment.py
----------------------------------------------------------------------
diff --git a/ambari-agent/src/main/python/resource_management/environment.py b/ambari-agent/src/main/python/resource_management/environment.py
deleted file mode 100644
index c9cb6e0..0000000
--- a/ambari-agent/src/main/python/resource_management/environment.py
+++ /dev/null
@@ -1,239 +0,0 @@
-#!/usr/bin/env python
-
-__all__ = ["Environment"]
-
-import logging
-import os
-import shutil
-import time
-from datetime import datetime
-
-from resource_management import shell
-from resource_management.exceptions import Fail
-from resource_management.providers import find_provider
-from resource_management.utils import AttributeDictionary, ParamsAttributeDictionary
-from resource_management.system import System
-from string import Template
-
-
-class Environment(object):
- _instances = []
-
- def __init__(self, basedir=None, params=None):
- """
- @param basedir: basedir/files, basedir/templates are the places where templates / static files
- are looked up
- @param params: configurations dictionary (this will be accessible in the templates)
- """
- self.log = logging.getLogger("resource_management")
- self.reset(basedir, params)
-
- def reset(self, basedir, params):
- self.system = System.get_instance()
- self.config = AttributeDictionary()
- self.resources = {}
- self.resource_list = []
- Substitutor.default_prefixes = []
- self.delayed_actions = set()
- self.update_config({
- # current time
- 'date': datetime.now(),
- # backups here files which were rewritten while executing File resource
- 'backup.path': '/tmp/resource_management/backup',
- # prefix for this files
- 'backup.prefix': datetime.now().strftime("%Y%m%d%H%M%S"),
- # dir where templates,failes dirs are
- 'basedir': basedir,
- # variables, which can be used in templates
- 'params': ParamsAttributeDictionary(Substitutor, params),
- })
-
- def backup_file(self, path):
- if self.config.backup:
- if not os.path.exists(self.config.backup.path):
- os.makedirs(self.config.backup.path, 0700)
- new_name = self.config.backup.prefix + path.replace('/', '-')
- backup_path = os.path.join(self.config.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 set_default_prefixes(self, dict):
- Substitutor.default_prefixes = dict
-
- def _check_condition(self, cond):
- if hasattr(cond, '__call__'):
- return cond()
-
- if isinstance(cond, basestring):
- ret, out = shell.call(cond)
- 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.initial_wait:
- time.sleep(resource.initial_wait)
-
- 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:
- if not resource.ignore_failures:
- self.run_action(resource, action)
- else:
- try:
- self.run_action(resource, action)
- except Exception:
- pass
-
- # 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']
-
-
-class Substitutor():
- log = logging.getLogger("resource_management.resource")
- default_prefixes = []
-
- class ExtendedTemplate(Template):
- """
- This is done to support substitution of dictionaries in dictionaries
- ( ':' sign)
-
- default is:
- idpattern = r'[_a-z][_a-z0-9]*'
- """
- idpattern = r'[_a-z][_a-z0-9:]*'
-
- @staticmethod
- def _get_subdict(name, dic):
- """
- "a:b:c" => a[b][c]
-
- doesn't use prefixes
- """
- name_parts = name.split(':')
- curr = dic
-
- for x in name_parts:
- curr = curr[x]
- return curr
-
- @staticmethod
- def get_subdict(name, dic):
- """
- "a:b:c" => a[b][c]
-
- can use prefixes
- """
- prefixes = list(Substitutor.default_prefixes)
- prefixes.insert(0, None) # for not prefixed case
- name_parts = name.split(':')
- is_found = False
- result = None
-
- for prefix in prefixes:
- curr = Substitutor._get_subdict(prefix,dic) if prefix else dic
-
- try:
- for x in name_parts:
- curr = curr[x]
- except (KeyError, TypeError):
- continue
-
- if is_found:
- raise Fail("Variable ${%s} found more than one time, please check your default prefixes!" % name)
-
- is_found = True
- result = curr
-
- if not result:
- raise Fail("Configuration on ${%s} cannot be resolved" % name)
-
- return result
-
- @staticmethod
- def substitute(val):
- env = Environment.get_instance()
- dic = env.config.params
-
- if dic and isinstance(val, str):
- result = Substitutor.ExtendedTemplate(val).substitute(dic)
- if '$' in val:
- Substitutor.log.debug("%s after substitution is %s", val, result)
- return result
-
- return val
http://git-wip-us.apache.org/repos/asf/incubator-ambari/blob/84274b4f/ambari-agent/src/main/python/resource_management/exceptions.py
----------------------------------------------------------------------
diff --git a/ambari-agent/src/main/python/resource_management/exceptions.py b/ambari-agent/src/main/python/resource_management/exceptions.py
deleted file mode 100644
index 26cd3a2..0000000
--- a/ambari-agent/src/main/python/resource_management/exceptions.py
+++ /dev/null
@@ -1,6 +0,0 @@
-class Fail(Exception):
- pass
-
-
-class InvalidArgument(Fail):
- pass
http://git-wip-us.apache.org/repos/asf/incubator-ambari/blob/84274b4f/ambari-agent/src/main/python/resource_management/libraries/__init__.py
----------------------------------------------------------------------
diff --git a/ambari-agent/src/main/python/resource_management/libraries/__init__.py b/ambari-agent/src/main/python/resource_management/libraries/__init__.py
new file mode 100644
index 0000000..cff8578
--- /dev/null
+++ b/ambari-agent/src/main/python/resource_management/libraries/__init__.py
@@ -0,0 +1,2 @@
+from resource_management.libraries.resources import *
+from resource_management.libraries.providers import *
http://git-wip-us.apache.org/repos/asf/incubator-ambari/blob/84274b4f/ambari-agent/src/main/python/resource_management/libraries/providers/__init__.py
----------------------------------------------------------------------
diff --git a/ambari-agent/src/main/python/resource_management/libraries/providers/__init__.py b/ambari-agent/src/main/python/resource_management/libraries/providers/__init__.py
new file mode 100644
index 0000000..1cb185c
--- /dev/null
+++ b/ambari-agent/src/main/python/resource_management/libraries/providers/__init__.py
@@ -0,0 +1,14 @@
+PROVIDERS = dict(
+ redhat=dict(
+ ),
+ centos=dict(
+ ),
+ suse=dict(
+ ),
+ fedora=dict(
+ ),
+ amazon=dict(
+ ),
+ default=dict(
+ ),
+)
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-ambari/blob/84274b4f/ambari-agent/src/main/python/resource_management/libraries/resources/__init__.py
----------------------------------------------------------------------
diff --git a/ambari-agent/src/main/python/resource_management/libraries/resources/__init__.py b/ambari-agent/src/main/python/resource_management/libraries/resources/__init__.py
new file mode 100644
index 0000000..e69de29