Skip to content
Branch: master
Find file Copy path
Find file Copy path
Fetching contributors…
Cannot retrieve contributors at this time
1491 lines (1197 sloc) 46.6 KB
# Copyright 2014-2015 Canonical Limited.
# Licensed 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
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# See the License for the specific language governing permissions and
# limitations under the License.
"Interactions with the Juju environment"
# Copyright 2013 Canonical Ltd.
# Authors:
# Charm Helpers Developers <>
from __future__ import print_function
import copy
from distutils.version import LooseVersion
from functools import wraps
from collections import namedtuple
import glob
import os
import json
import yaml
import re
import subprocess
import sys
import errno
import tempfile
from subprocess import CalledProcessError
import six
if not six.PY3:
from UserDict import UserDict
from collections import UserDict
MARKER = object()
SH_MAX_ARG = 131071
RANGE_WARNING = ('Passing NO_PROXY string that includes a cidr. '
'This may not be compatible with software you are '
'running in your shell.')
cache = {}
def cached(func):
"""Cache return values for multiple executions of func + args
For example::
def unit_get(attribute):
will cache the result of unit_get + 'test' for future calls.
def wrapper(*args, **kwargs):
global cache
key = json.dumps((func, args, kwargs), sort_keys=True, default=str)
return cache[key]
except KeyError:
pass # Drop out of the exception handler scope.
res = func(*args, **kwargs)
cache[key] = res
return res
wrapper._wrapped = func
return wrapper
def flush(key):
"""Flushes any entries from function cache where the
key is found in the function+args """
flush_list = []
for item in cache:
if key in item:
for item in flush_list:
del cache[item]
def log(message, level=None):
"""Write a message to the juju log"""
command = ['juju-log']
if level:
command += ['-l', level]
if not isinstance(message, six.string_types):
message = repr(message)
command += [message[:SH_MAX_ARG]]
# Missing juju-log should not cause failures in unit tests
# Send log output to stderr
except OSError as e:
if e.errno == errno.ENOENT:
if level:
message = "{}: {}".format(level, message)
message = "juju-log: {}".format(message)
print(message, file=sys.stderr)
class Serializable(UserDict):
"""Wrapper, an object that can be serialized to yaml or json"""
def __init__(self, obj):
# wrap the object
UserDict.__init__(self) = obj
def __getattr__(self, attr):
# See if this object has attribute.
if attr in ("json", "yaml", "data"):
return self.__dict__[attr]
# Check for attribute in wrapped object.
got = getattr(, attr, MARKER)
if got is not MARKER:
return got
# Proxy to the wrapped object via dict interface.
except KeyError:
raise AttributeError(attr)
def __getstate__(self):
# Pickle as a standard dictionary.
def __setstate__(self, state):
# Unpickle into our wrapper. = state
def json(self):
"""Serialize the object to json"""
return json.dumps(
def yaml(self):
"""Serialize the object to yaml"""
return yaml.dump(
def execution_environment():
"""A convenient bundling of the current execution context"""
context = {}
context['conf'] = config()
if relation_id():
context['reltype'] = relation_type()
context['relid'] = relation_id()
context['rel'] = relation_get()
context['unit'] = local_unit()
context['rels'] = relations()
context['env'] = os.environ
return context
def in_relation_hook():
"""Determine whether we're running in a relation hook"""
return 'JUJU_RELATION' in os.environ
def relation_type():
"""The scope for the current relation hook"""
return os.environ.get('JUJU_RELATION', None)
def relation_id(relation_name=None, service_or_unit=None):
"""The relation ID for the current or a specified relation"""
if not relation_name and not service_or_unit:
return os.environ.get('JUJU_RELATION_ID', None)
elif relation_name and service_or_unit:
service_name = service_or_unit.split('/')[0]
for relid in relation_ids(relation_name):
remote_service = remote_service_name(relid)
if remote_service == service_name:
return relid
raise ValueError('Must specify neither or both of relation_name and service_or_unit')
def local_unit():
"""Local unit ID"""
return os.environ['JUJU_UNIT_NAME']
def remote_unit():
"""The remote unit for the current relation hook"""
return os.environ.get('JUJU_REMOTE_UNIT', None)
def application_name():
The name of the deployed application this unit belongs to.
return local_unit().split('/')[0]
def service_name():
.. deprecated:: 0.19.1
Alias for :func:`application_name`.
return application_name()
def model_name():
Name of the model that this unit is deployed in.
return os.environ['JUJU_MODEL_NAME']
def model_uuid():
UUID of the model that this unit is deployed in.
return os.environ['JUJU_MODEL_UUID']
def principal_unit():
"""Returns the principal unit of this unit, otherwise None"""
# Juju 2.2 and above provides JUJU_PRINCIPAL_UNIT
principal_unit = os.environ.get('JUJU_PRINCIPAL_UNIT', None)
# If it's empty, then this unit is the principal
if principal_unit == '':
return os.environ['JUJU_UNIT_NAME']
elif principal_unit is not None:
return principal_unit
# For Juju 2.1 and below, let's try work out the principle unit by
# the various charms' metadata.yaml.
for reltype in relation_types():
for rid in relation_ids(reltype):
for unit in related_units(rid):
md = _metadata_unit(unit)
if not md:
subordinate = md.pop('subordinate', None)
if not subordinate:
return unit
return None
def remote_service_name(relid=None):
"""The remote service name for a given relation-id (or the current relation)"""
if relid is None:
unit = remote_unit()
units = related_units(relid)
unit = units[0] if units else None
return unit.split('/')[0] if unit else None
def hook_name():
"""The name of the currently executing hook"""
return os.environ.get('JUJU_HOOK_NAME', os.path.basename(sys.argv[0]))
class Config(dict):
"""A dictionary representation of the charm's config.yaml, with some
extra features:
- See which values in the dictionary have changed since the previous hook.
- For values that have changed, see what the previous value was.
- Store arbitrary data for use in a later hook.
NOTE: Do not instantiate this object directly - instead call
``hookenv.config()``, which will return an instance of :class:`Config`.
Example usage::
>>> # inside a hook
>>> from charmhelpers.core import hookenv
>>> config = hookenv.config()
>>> config['foo']
>>> # store a new key/value for later use
>>> config['mykey'] = 'myval'
>>> # user runs `juju set mycharm foo=baz`
>>> # now we're inside subsequent config-changed hook
>>> config = hookenv.config()
>>> config['foo']
>>> # test to see if this val has changed since last hook
>>> config.changed('foo')
>>> # what was the previous value?
>>> config.previous('foo')
>>> # keys/values that we add are preserved across hooks
>>> config['mykey']
CONFIG_FILE_NAME = '.juju-persistent-config'
def __init__(self, *args, **kw):
super(Config, self).__init__(*args, **kw)
self.implicit_save = True
self._prev_dict = None
self.path = os.path.join(charm_dir(), Config.CONFIG_FILE_NAME)
if os.path.exists(self.path) and os.stat(self.path).st_size:
def load_previous(self, path=None):
"""Load previous copy of config from disk.
In normal usage you don't need to call this method directly - it
is called automatically at object initialization.
:param path:
File path from which to load the previous config. If `None`,
config is loaded from the default location. If `path` is
specified, subsequent `save()` calls will write to the same
self.path = path or self.path
with open(self.path) as f:
self._prev_dict = json.load(f)
except ValueError as e:
log('Unable to parse previous config data - {}'.format(str(e)),
for k, v in copy.deepcopy(self._prev_dict).items():
if k not in self:
self[k] = v
def changed(self, key):
"""Return True if the current value for this key is different from
the previous value.
if self._prev_dict is None:
return True
return self.previous(key) != self.get(key)
def previous(self, key):
"""Return previous value for this key, or None if there
is no previous value.
if self._prev_dict:
return self._prev_dict.get(key)
return None
def save(self):
"""Save this config to disk.
If the charm is using the :mod:`Services Framework <services.base>`
or :meth:'@hook <Hooks.hook>' decorator, this
is called automatically at the end of successful hook execution.
Otherwise, it should be called directly by user code.
To disable automatic saves, set ``implicit_save=False`` on this
with open(self.path, 'w') as f:
os.fchmod(f.fileno(), 0o600)
json.dump(self, f)
def _implicit_save(self):
if self.implicit_save:
_cache_config = None
def config(scope=None):
Get the juju charm configuration (scope==None) or individual key,
(scope=str). The returned value is a Python data structure loaded as
JSON from the Juju config command.
:param scope: If set, return the value for the specified key.
:type scope: Optional[str]
:returns: Either the whole config as a Config, or a key from it.
:rtype: Any
global _cache_config
config_cmd_line = ['config-get', '--all', '--format=json']
# JSON Decode Exception for Python3.5+
exc_json = json.decoder.JSONDecodeError
except AttributeError:
# JSON Decode Exception for Python2.7 through Python3.4
exc_json = ValueError
if _cache_config is None:
config_data = json.loads(
_cache_config = Config(config_data)
if scope is not None:
return _cache_config.get(scope)
return _cache_config
except (exc_json, UnicodeDecodeError) as e:
log('Unable to parse output from config-get: config_cmd_line="{}" '
.format(config_cmd_line, str(e)), level=ERROR)
return None
def relation_get(attribute=None, unit=None, rid=None):
"""Get relation information"""
_args = ['relation-get', '--format=json']
if rid:
_args.append(attribute or '-')
if unit:
return json.loads(subprocess.check_output(_args).decode('UTF-8'))
except ValueError:
return None
except CalledProcessError as e:
if e.returncode == 2:
return None
def relation_set(relation_id=None, relation_settings=None, **kwargs):
"""Set relation information for the current unit"""
relation_settings = relation_settings if relation_settings else {}
relation_cmd_line = ['relation-set']
accepts_file = "--file" in subprocess.check_output(
relation_cmd_line + ["--help"], universal_newlines=True)
if relation_id is not None:
relation_cmd_line.extend(('-r', relation_id))
settings = relation_settings.copy()
for key, value in settings.items():
# Force value to be a string: it always should, but some call
# sites pass in things like dicts or numbers.
if value is not None:
settings[key] = "{}".format(value)
if accepts_file:
# --file was introduced in Juju 1.23.2. Use it by default if
# available, since otherwise we'll break if the relation data is
# too big. Ideally we should tell relation-set to read the data from
# stdin, but that feature is broken in 1.23.2: Bug #1454678.
with tempfile.NamedTemporaryFile(delete=False) as settings_file:
relation_cmd_line + ["--file",])
for key, value in settings.items():
if value is None:
relation_cmd_line.append('{}={}'.format(key, value))
# Flush cache of any relation-gets for local unit
def relation_clear(r_id=None):
''' Clears any relation data already set on relation r_id '''
settings = relation_get(rid=r_id,
for setting in settings:
if setting not in ['public-address', 'private-address']:
settings[setting] = None
def relation_ids(reltype=None):
"""A list of relation_ids"""
reltype = reltype or relation_type()
relid_cmd_line = ['relation-ids', '--format=json']
if reltype is not None:
return json.loads(
subprocess.check_output(relid_cmd_line).decode('UTF-8')) or []
return []
def related_units(relid=None):
"""A list of related units"""
relid = relid or relation_id()
units_cmd_line = ['relation-list', '--format=json']
if relid is not None:
units_cmd_line.extend(('-r', relid))
return json.loads(
subprocess.check_output(units_cmd_line).decode('UTF-8')) or []
def expected_peer_units():
"""Get a generator for units we expect to join peer relation based on
The local unit is excluded from the result to make it easy to gauge
completion of all peers joining the relation with existing hook tools.
Example usage:
log('peer {} of {} joined peer relation'
This function will raise NotImplementedError if used with juju versions
without goal-state support.
:returns: iterator
:rtype: types.GeneratorType
:raises: NotImplementedError
if not has_juju_version("2.4.0"):
# goal-state first appeared in 2.4.0.
raise NotImplementedError("goal-state")
_goal_state = goal_state()
return (key for key in _goal_state['units']
if '/' in key and key != local_unit())
def expected_related_units(reltype=None):
"""Get a generator for units we expect to join relation based on
Note that you can not use this function for the peer relation, take a look
at expected_peer_units() for that.
This function will raise KeyError if you request information for a
relation type for which juju goal-state does not have information. It will
raise NotImplementedError if used with juju versions without goal-state
Example usage:
log('participant {} of {} joined relation {}'
:param reltype: Relation type to list data for, default is to list data for
the realtion type we are currently executing a hook for.
:type reltype: str
:returns: iterator
:rtype: types.GeneratorType
:raises: KeyError, NotImplementedError
if not has_juju_version("2.4.4"):
# goal-state existed in 2.4.0, but did not list individual units to
# join a relation in 2.4.1 through 2.4.3. (LP: #1794739)
raise NotImplementedError("goal-state relation unit count")
reltype = reltype or relation_type()
_goal_state = goal_state()
return (key for key in _goal_state['relations'][reltype] if '/' in key)
def relation_for_unit(unit=None, rid=None):
"""Get the json represenation of a unit's relation"""
unit = unit or remote_unit()
relation = relation_get(unit=unit, rid=rid)
for key in relation:
if key.endswith('-list'):
relation[key] = relation[key].split()
relation['__unit__'] = unit
return relation
def relations_for_id(relid=None):
"""Get relations of a specific relation ID"""
relation_data = []
relid = relid or relation_ids()
for unit in related_units(relid):
unit_data = relation_for_unit(unit, relid)
unit_data['__relid__'] = relid
return relation_data
def relations_of_type(reltype=None):
"""Get relations of a specific type"""
relation_data = []
reltype = reltype or relation_type()
for relid in relation_ids(reltype):
for relation in relations_for_id(relid):
relation['__relid__'] = relid
return relation_data
def metadata():
"""Get the current charm metadata.yaml contents as a python object"""
with open(os.path.join(charm_dir(), 'metadata.yaml')) as md:
return yaml.safe_load(md)
def _metadata_unit(unit):
"""Given the name of a unit (e.g. apache2/0), get the unit charm's
metadata.yaml. Very similar to metadata() but allows us to inspect
other units. Unit needs to be co-located, such as a subordinate or
:returns: metadata.yaml as a python object.
basedir = os.sep.join(charm_dir().split(os.sep)[:-2])
unitdir = 'unit-{}'.format(unit.replace(os.sep, '-'))
joineddir = os.path.join(basedir, unitdir, 'charm', 'metadata.yaml')
if not os.path.exists(joineddir):
return None
with open(joineddir) as md:
return yaml.safe_load(md)
def relation_types():
"""Get a list of relation types supported by this charm"""
rel_types = []
md = metadata()
for key in ('provides', 'requires', 'peers'):
section = md.get(key)
if section:
return rel_types
def peer_relation_id():
'''Get the peers relation id if a peers relation has been joined, else None.'''
md = metadata()
section = md.get('peers')
if section:
for key in section:
relids = relation_ids(key)
if relids:
return relids[0]
return None
def relation_to_interface(relation_name):
Given the name of a relation, return the interface that relation uses.
:returns: The interface name, or ``None``.
return relation_to_role_and_interface(relation_name)[1]
def relation_to_role_and_interface(relation_name):
Given the name of a relation, return the role and the name of the interface
that relation uses (where role is one of ``provides``, ``requires``, or ``peers``).
:returns: A tuple containing ``(role, interface)``, or ``(None, None)``.
_metadata = metadata()
for role in ('provides', 'requires', 'peers'):
interface = _metadata.get(role, {}).get(relation_name, {}).get('interface')
if interface:
return role, interface
return None, None
def role_and_interface_to_relations(role, interface_name):
Given a role and interface name, return a list of relation names for the
current charm that use that interface under that role (where role is one
of ``provides``, ``requires``, or ``peers``).
:returns: A list of relation names.
_metadata = metadata()
results = []
for relation_name, relation in _metadata.get(role, {}).items():
if relation['interface'] == interface_name:
return results
def interface_to_relations(interface_name):
Given an interface, return a list of relation names for the current
charm that use that interface.
:returns: A list of relation names.
results = []
for role in ('provides', 'requires', 'peers'):
results.extend(role_and_interface_to_relations(role, interface_name))
return results
def charm_name():
"""Get the name of the current charm as is specified on metadata.yaml"""
return metadata().get('name')
def relations():
"""Get a nested dictionary of relation data for all related units"""
rels = {}
for reltype in relation_types():
relids = {}
for relid in relation_ids(reltype):
units = {local_unit(): relation_get(unit=local_unit(), rid=relid)}
for unit in related_units(relid):
reldata = relation_get(unit=unit, rid=relid)
units[unit] = reldata
relids[relid] = units
rels[reltype] = relids
return rels
def is_relation_made(relation, keys='private-address'):
Determine whether a relation is established by checking for
presence of key(s). If a list of keys is provided, they
must all be present for the relation to be identified as made
if isinstance(keys, str):
keys = [keys]
for r_id in relation_ids(relation):
for unit in related_units(r_id):
context = {}
for k in keys:
context[k] = relation_get(k, rid=r_id,
if None not in context.values():
return True
return False
def _port_op(op_name, port, protocol="TCP"):
"""Open or close a service network port"""
_args = [op_name]
icmp = protocol.upper() == "ICMP"
if icmp:
_args.append('{}/{}'.format(port, protocol))
except subprocess.CalledProcessError:
# Older Juju pre 2.3 doesn't support ICMP
# so treat it as a no-op if it fails.
if not icmp:
def open_port(port, protocol="TCP"):
"""Open a service network port"""
_port_op('open-port', port, protocol)
def close_port(port, protocol="TCP"):
"""Close a service network port"""
_port_op('close-port', port, protocol)
def open_ports(start, end, protocol="TCP"):
"""Opens a range of service network ports"""
_args = ['open-port']
_args.append('{}-{}/{}'.format(start, end, protocol))
def close_ports(start, end, protocol="TCP"):
"""Close a range of service network ports"""
_args = ['close-port']
_args.append('{}-{}/{}'.format(start, end, protocol))
def opened_ports():
"""Get the opened ports
*Note that this will only show ports opened in a previous hook*
:returns: Opened ports as a list of strings: ``['8080/tcp', '8081-8083/tcp']``
_args = ['opened-ports', '--format=json']
return json.loads(subprocess.check_output(_args).decode('UTF-8'))
def unit_get(attribute):
"""Get the unit ID for the remote unit"""
_args = ['unit-get', '--format=json', attribute]
return json.loads(subprocess.check_output(_args).decode('UTF-8'))
except ValueError:
return None
def unit_public_ip():
"""Get this unit's public IP address"""
return unit_get('public-address')
def unit_private_ip():
"""Get this unit's private IP address"""
return unit_get('private-address')
def storage_get(attribute=None, storage_id=None):
"""Get storage attributes"""
_args = ['storage-get', '--format=json']
if storage_id:
_args.extend(('-s', storage_id))
if attribute:
return json.loads(subprocess.check_output(_args).decode('UTF-8'))
except ValueError:
return None
def storage_list(storage_name=None):
"""List the storage IDs for the unit"""
_args = ['storage-list', '--format=json']
if storage_name:
return json.loads(subprocess.check_output(_args).decode('UTF-8'))
except ValueError:
return None
except OSError as e:
import errno
if e.errno == errno.ENOENT:
# storage-list does not exist
return []
class UnregisteredHookError(Exception):
"""Raised when an undefined hook is called"""
class Hooks(object):
"""A convenient handler for hook functions.
hooks = Hooks()
# register a hook, taking its name from the function name
def install():
pass # your code here
# register a hook, providing a custom hook name
def config_changed():
pass # your code here
if __name__ == "__main__":
# execute a hook based on the name the program is called by
def __init__(self, config_save=None):
super(Hooks, self).__init__()
self._hooks = {}
# For unknown reasons, we allow the Hooks constructor to override
# config().implicit_save.
if config_save is not None:
config().implicit_save = config_save
def register(self, name, function):
"""Register a hook"""
self._hooks[name] = function
def execute(self, args):
"""Execute a registered hook based on args[0]"""
hook_name = os.path.basename(args[0])
if hook_name in self._hooks:
except SystemExit as x:
if x.code is None or x.code == 0:
raise UnregisteredHookError(hook_name)
def hook(self, *hook_names):
"""Decorator, registering them as hooks"""
def wrapper(decorated):
for hook_name in hook_names:
self.register(hook_name, decorated)
self.register(decorated.__name__, decorated)
if '_' in decorated.__name__:
decorated.__name__.replace('_', '-'), decorated)
return decorated
return wrapper
class NoNetworkBinding(Exception):
def charm_dir():
"""Return the root directory of the current charm"""
d = os.environ.get('JUJU_CHARM_DIR')
if d is not None:
return d
return os.environ.get('CHARM_DIR')
def action_get(key=None):
"""Gets the value of an action parameter, or all key/value param pairs"""
cmd = ['action-get']
if key is not None:
action_data = json.loads(subprocess.check_output(cmd).decode('UTF-8'))
return action_data
def action_set(values):
"""Sets the values to be returned after the action finishes"""
cmd = ['action-set']
for k, v in list(values.items()):
cmd.append('{}={}'.format(k, v))
def action_fail(message):
"""Sets the action status to failed and sets the error message.
The results set by action_set are preserved."""
subprocess.check_call(['action-fail', message])
def action_name():
"""Get the name of the currently executing action."""
return os.environ.get('JUJU_ACTION_NAME')
def action_uuid():
"""Get the UUID of the currently executing action."""
return os.environ.get('JUJU_ACTION_UUID')
def action_tag():
"""Get the tag for the currently executing action."""
return os.environ.get('JUJU_ACTION_TAG')
def status_set(workload_state, message):
"""Set the workload state with a message
Use status-set to set the workload state with a message which is visible
to the user via juju status. If the status-set command is not found then
assume this is juju < 1.23 and juju-log the message unstead.
workload_state -- valid juju workload state.
message -- status update message
valid_states = ['maintenance', 'blocked', 'waiting', 'active']
if workload_state not in valid_states:
raise ValueError(
'{!r} is not a valid workload state'.format(workload_state)
cmd = ['status-set', workload_state, message]
ret =
if ret == 0:
except OSError as e:
if e.errno != errno.ENOENT:
log_message = 'status-set failed: {} {}'.format(workload_state,
log(log_message, level='INFO')
def status_get():
"""Retrieve the previously set juju workload state and message
If the status-get command is not found then assume this is juju < 1.23 and
return 'unknown', ""
cmd = ['status-get', "--format=json", "--include-data"]
raw_status = subprocess.check_output(cmd)
except OSError as e:
if e.errno == errno.ENOENT:
return ('unknown', "")
status = json.loads(raw_status.decode("UTF-8"))
return (status["status"], status["message"])
def translate_exc(from_exc, to_exc):
def inner_translate_exc1(f):
def inner_translate_exc2(*args, **kwargs):
return f(*args, **kwargs)
except from_exc:
raise to_exc
return inner_translate_exc2
return inner_translate_exc1
def application_version_set(version):
"""Charm authors may trigger this command from any hook to output what
version of the application is running. This could be a package version,
for instance postgres version 9.5. It could also be a build number or
version control revision identifier, for instance git sha 6fb7ba68. """
cmd = ['application-version-set']
except OSError:
log("Application Version: {}".format(version))
@translate_exc(from_exc=OSError, to_exc=NotImplementedError)
def goal_state():
"""Juju goal state values"""
cmd = ['goal-state', '--format=json']
return json.loads(subprocess.check_output(cmd).decode('UTF-8'))
@translate_exc(from_exc=OSError, to_exc=NotImplementedError)
def is_leader():
"""Does the current unit hold the juju leadership
Uses juju to determine whether the current unit is the leader of its peers
cmd = ['is-leader', '--format=json']
return json.loads(subprocess.check_output(cmd).decode('UTF-8'))
@translate_exc(from_exc=OSError, to_exc=NotImplementedError)
def leader_get(attribute=None):
"""Juju leader get value(s)"""
cmd = ['leader-get', '--format=json'] + [attribute or '-']
return json.loads(subprocess.check_output(cmd).decode('UTF-8'))
@translate_exc(from_exc=OSError, to_exc=NotImplementedError)
def leader_set(settings=None, **kwargs):
"""Juju leader set value(s)"""
# Don't log secrets.
# log("Juju leader-set '%s'" % (settings), level=DEBUG)
cmd = ['leader-set']
settings = settings or {}
for k, v in settings.items():
if v is None:
cmd.append('{}={}'.format(k, v))
@translate_exc(from_exc=OSError, to_exc=NotImplementedError)
def payload_register(ptype, klass, pid):
""" is used while a hook is running to let Juju know that a
payload has been started."""
cmd = ['payload-register']
for x in [ptype, klass, pid]:
@translate_exc(from_exc=OSError, to_exc=NotImplementedError)
def payload_unregister(klass, pid):
""" is used while a hook is running to let Juju know
that a payload has been manually stopped. The <class> and <id> provided
must match a payload that has been previously registered with juju using
cmd = ['payload-unregister']
for x in [klass, pid]:
@translate_exc(from_exc=OSError, to_exc=NotImplementedError)
def payload_status_set(klass, pid, status):
"""is used to update the current status of a registered payload.
The <class> and <id> provided must match a payload that has been previously
registered with juju using payload-register. The <status> must be one of the
follow: starting, started, stopping, stopped"""
cmd = ['payload-status-set']
for x in [klass, pid, status]:
@translate_exc(from_exc=OSError, to_exc=NotImplementedError)
def resource_get(name):
"""used to fetch the resource path of the given name.
<name> must match a name of defined resource in metadata.yaml
returns either a path or False if resource not available
if not name:
return False
cmd = ['resource-get', name]
return subprocess.check_output(cmd).decode('UTF-8')
except subprocess.CalledProcessError:
return False
def juju_version():
"""Full version string (eg. '')"""
# Per
jujud = glob.glob('/var/lib/juju/tools/machine-*/jujud')[0]
return subprocess.check_output([jujud, 'version'],
def has_juju_version(minimum_version):
"""Return True if the Juju version is at least the provided version"""
return LooseVersion(juju_version()) >= LooseVersion(minimum_version)
_atexit = []
_atstart = []
def atstart(callback, *args, **kwargs):
'''Schedule a callback to run before the main hook.
Callbacks are run in the order they were added.
This is useful for modules and classes to perform initialization
and inject behavior. In particular:
- Run common code before all of your hooks, such as logging
the hook name or interesting relation data.
- Defer object or module initialization that requires a hook
context until we know there actually is a hook context,
making testing easier.
- Rather than requiring charm authors to include boilerplate to
invoke your helper's behavior, have it run automatically if
your object is instantiated or module imported.
This is not at all useful after your hook framework as been launched.
global _atstart
_atstart.append((callback, args, kwargs))
def atexit(callback, *args, **kwargs):
'''Schedule a callback to run on successful hook completion.
Callbacks are run in the reverse order that they were added.'''
_atexit.append((callback, args, kwargs))
def _run_atstart():
'''Hook frameworks must invoke this before running the main hook body.'''
global _atstart
for callback, args, kwargs in _atstart:
callback(*args, **kwargs)
del _atstart[:]
def _run_atexit():
'''Hook frameworks must invoke this after the main hook body has
successfully completed. Do not invoke it if the hook fails.'''
global _atexit
for callback, args, kwargs in reversed(_atexit):
callback(*args, **kwargs)
del _atexit[:]
@translate_exc(from_exc=OSError, to_exc=NotImplementedError)
def network_get_primary_address(binding):
Deprecated since Juju 2.3; use network_get()
Retrieve the primary network address for a named binding
:param binding: string. The name of a relation of extra-binding
:return: string. The primary IP address for the named binding
:raise: NotImplementedError if run on Juju < 2.0
cmd = ['network-get', '--primary-address', binding]
response = subprocess.check_output(
except CalledProcessError as e:
if 'no network config found for binding' in e.output.decode('UTF-8'):
raise NoNetworkBinding("No network binding for {}"
return response
def network_get(endpoint, relation_id=None):
Retrieve the network details for a relation endpoint
:param endpoint: string. The name of a relation endpoint
:param relation_id: int. The ID of the relation for the current context.
:return: dict. The loaded YAML output of the network-get query.
:raise: NotImplementedError if request not supported by the Juju version.
if not has_juju_version('2.2'):
raise NotImplementedError(juju_version()) # earlier versions require --primary-address
if relation_id and not has_juju_version('2.3'):
raise NotImplementedError # 2.3 added the -r option
cmd = ['network-get', endpoint, '--format', 'yaml']
if relation_id:
response = subprocess.check_output(
return yaml.safe_load(response)
def add_metric(*args, **kwargs):
"""Add metric values. Values may be expressed with keyword arguments. For
metric names containing dashes, these may be expressed as one or more
'key=value' positional arguments. May only be called from the collect-metrics
_args = ['add-metric']
_kvpairs = []
_kvpairs.extend(['{}={}'.format(k, v) for k, v in kwargs.items()])
except EnvironmentError as e:
if e.errno != errno.ENOENT:
log_message = 'add-metric failed: {}'.format(' '.join(_kvpairs))
log(log_message, level='INFO')
def meter_status():
"""Get the meter status, if running in the meter-status-changed hook."""
return os.environ.get('JUJU_METER_STATUS')
def meter_info():
"""Get the meter status information, if running in the meter-status-changed
return os.environ.get('JUJU_METER_INFO')
def iter_units_for_relation_name(relation_name):
"""Iterate through all units in a relation
Generator that iterates through all the units in a relation and yields
a named tuple with rid and unit field names.
data = [(u.rid, u.unit)
for u in iter_units_for_relation_name(relation_name)]
:param relation_name: string relation name
:yield: Named Tuple with rid and unit field names
RelatedUnit = namedtuple('RelatedUnit', 'rid, unit')
for rid in relation_ids(relation_name):
for unit in related_units(rid):
yield RelatedUnit(rid, unit)
def ingress_address(rid=None, unit=None):
Retrieve the ingress-address from a relation when available.
Otherwise, return the private-address.
When used on the consuming side of the relation (unit is a remote
unit), the ingress-address is the IP address that this unit needs
to use to reach the provided service on the remote unit.
When used on the providing side of the relation (unit == local_unit()),
the ingress-address is the IP address that is advertised to remote
units on this relation. Remote units need to use this address to
reach the local provided service on this unit.
Note that charms may document some other method to use in
preference to the ingress_address(), such as an address provided
on a different relation attribute or a service discovery mechanism.
This allows charms to redirect inbound connections to their peers
or different applications such as load balancers.
addresses = [ingress_address(rid=u.rid, unit=u.unit)
for u in iter_units_for_relation_name(relation_name)]
:param rid: string relation id
:param unit: string unit name
:side effect: calls relation_get
:return: string IP address
settings = relation_get(rid=rid, unit=unit)
return (settings.get('ingress-address') or
def egress_subnets(rid=None, unit=None):
Retrieve the egress-subnets from a relation.
This function is to be used on the providing side of the
relation, and provides the ranges of addresses that client
connections may come from. The result is uninteresting on
the consuming side of a relation (unit == local_unit()).
Returns a stable list of subnets in CIDR format.
eg. ['', '2001::F00F/128']
If egress-subnets is not available, falls back to using the published
ingress-address, or finally private-address.
:param rid: string relation id
:param unit: string unit name
:side effect: calls relation_get
:return: list of subnets in CIDR format. eg. ['', '2001::F00F/128']
def _to_range(addr):
if'^(?:\d{1,3}\.){3}\d{1,3}$', addr) is not None:
addr += '/32'
elif ':' in addr and '/' not in addr: # IPv6
addr += '/128'
return addr
settings = relation_get(rid=rid, unit=unit)
if 'egress-subnets' in settings:
return [n.strip() for n in settings['egress-subnets'].split(',') if n.strip()]
if 'ingress-address' in settings:
return [_to_range(settings['ingress-address'])]
if 'private-address' in settings:
return [_to_range(settings['private-address'])]
return [] # Should never happen
def unit_doomed(unit=None):
"""Determines if the unit is being removed from the model
Requires Juju 2.4.1.
:param unit: string unit name, defaults to local_unit
:side effect: calls goal_state
:side effect: calls local_unit
:side effect: calls has_juju_version
:return: True if the unit is being removed, already gone, or never existed
if not has_juju_version("2.4.1"):
# We cannot risk blindly returning False for 'we don't know',
# because that could cause data loss; if call sites don't
# need an accurate answer, they likely don't need this helper
# at all.
# goal-state existed in 2.4.0, but did not handle removals
# correctly until 2.4.1.
raise NotImplementedError("is_doomed")
if unit is None:
unit = local_unit()
gs = goal_state()
units = gs.get('units', {})
if unit not in units:
return True
# I don't think 'dead' units ever show up in the goal-state, but
# check anyway in addition to 'dying'.
return units[unit]['status'] in ('dying', 'dead')
def env_proxy_settings(selected_settings=None):
"""Get proxy settings from process environment variables.
Get charm proxy settings from environment variables that correspond to
juju-http-proxy, juju-https-proxy and juju-no-proxy (available as of 2.4.2,
see lp:1782236) in a format suitable for passing to an application that
reacts to proxy settings passed as environment variables. Some applications
support lowercase or uppercase notation (e.g. curl), some support only
lowercase (e.g. wget), there are also subjectively rare cases of only
uppercase notation support. no_proxy CIDR and wildcard support also varies
between runtimes and applications as there is no enforced standard.
Some applications may connect to multiple destinations and expose config
options that would affect only proxy settings for a specific destination
these should be handled in charms in an application-specific manner.
:param selected_settings: format only a subset of possible settings
:type selected_settings: list
:rtype: Option(None, dict[str, str])
'http': 'HTTP_PROXY',
'https': 'HTTPS_PROXY',
'no_proxy': 'NO_PROXY',
'ftp': 'FTP_PROXY'
if selected_settings is None:
selected_settings = SUPPORTED_SETTINGS
selected_vars = [v for k, v in SUPPORTED_SETTINGS.items()
if k in selected_settings]
proxy_settings = {}
for var in selected_vars:
var_val = os.getenv(var)
if var_val:
proxy_settings[var] = var_val
proxy_settings[var.lower()] = var_val
# Now handle juju-prefixed environment variables. The legacy vs new
# environment variable usage is mutually exclusive
charm_var_val = os.getenv('JUJU_CHARM_{}'.format(var))
if charm_var_val:
proxy_settings[var] = charm_var_val
proxy_settings[var.lower()] = charm_var_val
if 'no_proxy' in proxy_settings:
if _contains_range(proxy_settings['no_proxy']):
return proxy_settings if proxy_settings else None
def _contains_range(addresses):
"""Check for cidr or wildcard domain in a string.
Given a string comprising a comma seperated list of ip addresses
and domain names, determine whether the string contains IP ranges
or wildcard domains.
:param addresses: comma seperated list of domains and ip addresses.
:type addresses: str
return (
# Test for cidr (e.g.
"/" in addresses or
# Test for wildcard domains (* or
"*" in addresses or
addresses.startswith(".") or
",." in addresses or
" ." in addresses)
You can’t perform that action at this time.