openstack/charm-hacluster
Code Issues Proposed changes

Redux to use charm-helpers

Browse Source
This commit is contained in:
James Page 2014-04-11 12:02:09 +01:00
parent f1c107ee21
commit 0f0c49ff7b
20 changed files with 1592 additions and 513 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
charm-helpers.yaml Normal file
View File

@ -0,0 +1,8 @@
branch: lp:charm-helpers
destination: hooks/charmhelpers
include:
- core
- fetch
- contrib.storage.linux
- contrib.hahelpers
- contrib.storage

0
hooks/lib/__init__.py → hooks/charmhelpers/__init__.py
View File

0
hooks/charmhelpers/contrib/__init__.py Normal file
View File

0
hooks/charmhelpers/contrib/hahelpers/__init__.py Normal file
View File

59
hooks/charmhelpers/contrib/hahelpers/apache.py Normal file
View File

@ -0,0 +1,59 @@
#
# Copyright 2012 Canonical Ltd.
#
# This file is sourced from lp:openstack-charm-helpers
#
# Authors:
# James Page <james.page@ubuntu.com>
# Adam Gandelman <adamg@ubuntu.com>
#
import subprocess
from charmhelpers.core.hookenv import (
config as config_get,
relation_get,
relation_ids,
related_units as relation_list,
log,
INFO,
)
def get_cert():
cert = config_get('ssl_cert')
key = config_get('ssl_key')
if not (cert and key):
log("Inspecting identity-service relations for SSL certificate.",
level=INFO)
cert = key = None
for r_id in relation_ids('identity-service'):
for unit in relation_list(r_id):
if not cert:
cert = relation_get('ssl_cert',
rid=r_id, unit=unit)
if not key:
key = relation_get('ssl_key',
rid=r_id, unit=unit)
return (cert, key)
def get_ca_cert():
ca_cert = config_get('ssl_ca')
if ca_cert is None:
log("Inspecting identity-service relations for CA SSL certificate.",
level=INFO)
for r_id in relation_ids('identity-service'):
for unit in relation_list(r_id):
if ca_cert is None:
ca_cert = relation_get('ca_cert',
rid=r_id, unit=unit)
return ca_cert
def install_ca_cert(ca_cert):
if ca_cert:
with open('/usr/local/share/ca-certificates/keystone_juju_ca_cert.crt',
'w') as crt:
crt.write(ca_cert)
subprocess.check_call(['update-ca-certificates', '--fresh'])

93
hooks/lib/cluster_utils.py → hooks/charmhelpers/contrib/hahelpers/cluster.py
View File

@ -1,24 +1,31 @@
#
# Copyright 2012 Canonical Ltd.
#
# This file is sourced from lp:openstack-charm-helpers
#
# Authors:
# James Page <james.page@ubuntu.com>
# Adam Gandelman <adamg@ubuntu.com>
#
from lib.utils import (
juju_log,
relation_ids,
relation_list,
relation_get,
get_unit_hostname,
config_get
)
import subprocess
import os
from socket import gethostname as get_unit_hostname
from charmhelpers.core.hookenv import (
log,
relation_ids,
related_units as relation_list,
relation_get,
config as config_get,
INFO,
ERROR,
unit_get,
)
class HAIncompleteConfig(Exception):
pass
def is_clustered():
for r_id in (relation_ids('ha') or []):
@ -67,12 +74,12 @@ def oldest_peer(peers):
def eligible_leader(resource):
if is_clustered():
if not is_leader(resource):
juju_log('INFO', 'Deferring action to CRM leader.')
log('Deferring action to CRM leader.', level=INFO)
return False
else:
peers = peer_units()
if peers and not oldest_peer(peers):
juju_log('INFO', 'Deferring action to oldest service unit.')
log('Deferring action to oldest service unit.', level=INFO)
return False
return True
@ -90,10 +97,14 @@ def https():
return True
for r_id in relation_ids('identity-service'):
for unit in relation_list(r_id):
if (relation_get('https_keystone', rid=r_id, unit=unit) and
relation_get('ssl_cert', rid=r_id, unit=unit) and
relation_get('ssl_key', rid=r_id, unit=unit) and
relation_get('ca_cert', rid=r_id, unit=unit)):
rel_state = [
relation_get('https_keystone', rid=r_id, unit=unit),
relation_get('ssl_cert', rid=r_id, unit=unit),
relation_get('ssl_key', rid=r_id, unit=unit),
relation_get('ca_cert', rid=r_id, unit=unit),
]
# NOTE: works around (LP: #1203241)
if (None not in rel_state) and ('' not in rel_state):
return True
return False
@ -115,16 +126,58 @@ def determine_api_port(public_port):
return public_port - (i * 10)
def determine_haproxy_port(public_port):
def determine_apache_port(public_port):
'''
Description: Determine correct proxy listening port based on public IP +
existence of HTTPS reverse proxy.
Description: Determine correct apache listening port based on public IP +
state of the cluster.
public_port: int: standard public port for given service
returns: int: the correct listening port for the HAProxy service
'''
i = 0
if https():
if len(peer_units()) > 0 or is_clustered():
i += 1
return public_port - (i * 10)
def get_hacluster_config():
'''
Obtains all relevant configuration from charm configuration required
for initiating a relation to hacluster:
ha-bindiface, ha-mcastport, vip, vip_iface, vip_cidr
returns: dict: A dict containing settings keyed by setting name.
raises: HAIncompleteConfig if settings are missing.
'''
settings = ['ha-bindiface', 'ha-mcastport', 'vip', 'vip_iface', 'vip_cidr']
conf = {}
for setting in settings:
conf[setting] = config_get(setting)
missing = []
[missing.append(s) for s, v in conf.iteritems() if v is None]
if missing:
log('Insufficient config data to configure hacluster.', level=ERROR)
raise HAIncompleteConfig
return conf
def canonical_url(configs, vip_setting='vip'):
'''
Returns the correct HTTP URL to this host given the state of HTTPS
configuration and hacluster.
:configs : OSTemplateRenderer: A config tempating object to inspect for
a complete https context.
:vip_setting: str: Setting in charm config that specifies
VIP address.
'''
scheme = 'http'
if 'https' in configs.complete_contexts():
scheme = 'https'
if is_clustered():
addr = config_get(vip_setting)
else:
addr = unit_get('private-address')
return '%s://%s' % (scheme, addr)

0
hooks/charmhelpers/contrib/storage/__init__.py Normal file
View File

0
hooks/charmhelpers/contrib/storage/linux/__init__.py Normal file
View File

387
hooks/charmhelpers/contrib/storage/linux/ceph.py Normal file
View File

@ -0,0 +1,387 @@
#
# Copyright 2012 Canonical Ltd.
#
# This file is sourced from lp:openstack-charm-helpers
#
# Authors:
# James Page <james.page@ubuntu.com>
# Adam Gandelman <adamg@ubuntu.com>
#
import os
import shutil
import json
import time
from subprocess import (
check_call,
check_output,
CalledProcessError
)
from charmhelpers.core.hookenv import (
relation_get,
relation_ids,
related_units,
log,
INFO,
WARNING,
ERROR
)
from charmhelpers.core.host import (
mount,
mounts,
service_start,
service_stop,
service_running,
umount,
)
from charmhelpers.fetch import (
apt_install,
)
KEYRING = '/etc/ceph/ceph.client.{}.keyring'
KEYFILE = '/etc/ceph/ceph.client.{}.key'
CEPH_CONF = """[global]
auth supported = {auth}
keyring = {keyring}
mon host = {mon_hosts}
log to syslog = {use_syslog}
err to syslog = {use_syslog}
clog to syslog = {use_syslog}
"""
def install():
''' Basic Ceph client installation '''
ceph_dir = "/etc/ceph"
if not os.path.exists(ceph_dir):
os.mkdir(ceph_dir)
apt_install('ceph-common', fatal=True)
def rbd_exists(service, pool, rbd_img):
''' Check to see if a RADOS block device exists '''
try:
out = check_output(['rbd', 'list', '--id', service,
'--pool', pool])
except CalledProcessError:
return False
else:
return rbd_img in out
def create_rbd_image(service, pool, image, sizemb):
''' Create a new RADOS block device '''
cmd = [
'rbd',
'create',
image,
'--size',
str(sizemb),
'--id',
service,
'--pool',
pool
]
check_call(cmd)
def pool_exists(service, name):
''' Check to see if a RADOS pool already exists '''
try:
out = check_output(['rados', '--id', service, 'lspools'])
except CalledProcessError:
return False
else:
return name in out
def get_osds(service):
'''
Return a list of all Ceph Object Storage Daemons
currently in the cluster
'''
version = ceph_version()
if version and version >= '0.56':
return json.loads(check_output(['ceph', '--id', service,
'osd', 'ls', '--format=json']))
else:
return None
def create_pool(service, name, replicas=2):
''' Create a new RADOS pool '''
if pool_exists(service, name):
log("Ceph pool {} already exists, skipping creation".format(name),
level=WARNING)
return
# Calculate the number of placement groups based
# on upstream recommended best practices.
osds = get_osds(service)
if osds:
pgnum = (len(osds) * 100 / replicas)
else:
# NOTE(james-page): Default to 200 for older ceph versions
# which don't support OSD query from cli
pgnum = 200
cmd = [
'ceph', '--id', service,
'osd', 'pool', 'create',
name, str(pgnum)
]
check_call(cmd)
cmd = [
'ceph', '--id', service,
'osd', 'pool', 'set', name,
'size', str(replicas)
]
check_call(cmd)
def delete_pool(service, name):
''' Delete a RADOS pool from ceph '''
cmd = [
'ceph', '--id', service,
'osd', 'pool', 'delete',
name, '--yes-i-really-really-mean-it'
]
check_call(cmd)
def _keyfile_path(service):
return KEYFILE.format(service)
def _keyring_path(service):
return KEYRING.format(service)
def create_keyring(service, key):
''' Create a new Ceph keyring containing key'''
keyring = _keyring_path(service)
if os.path.exists(keyring):
log('ceph: Keyring exists at %s.' % keyring, level=WARNING)
return
cmd = [
'ceph-authtool',
keyring,
'--create-keyring',
'--name=client.{}'.format(service),
'--add-key={}'.format(key)
]
check_call(cmd)
log('ceph: Created new ring at %s.' % keyring, level=INFO)
def create_key_file(service, key):
''' Create a file containing key '''
keyfile = _keyfile_path(service)
if os.path.exists(keyfile):
log('ceph: Keyfile exists at %s.' % keyfile, level=WARNING)
return
with open(keyfile, 'w') as fd:
fd.write(key)
log('ceph: Created new keyfile at %s.' % keyfile, level=INFO)
def get_ceph_nodes():
''' Query named relation 'ceph' to detemine current nodes '''
hosts = []
for r_id in relation_ids('ceph'):
for unit in related_units(r_id):
hosts.append(relation_get('private-address', unit=unit, rid=r_id))
return hosts
def configure(service, key, auth, use_syslog):
''' Perform basic configuration of Ceph '''
create_keyring(service, key)
create_key_file(service, key)
hosts = get_ceph_nodes()
with open('/etc/ceph/ceph.conf', 'w') as ceph_conf:
ceph_conf.write(CEPH_CONF.format(auth=auth,
keyring=_keyring_path(service),
mon_hosts=",".join(map(str, hosts)),
use_syslog=use_syslog))
modprobe('rbd')
def image_mapped(name):
''' Determine whether a RADOS block device is mapped locally '''
try:
out = check_output(['rbd', 'showmapped'])
except CalledProcessError:
return False
else:
return name in out
def map_block_storage(service, pool, image):
''' Map a RADOS block device for local use '''
cmd = [
'rbd',
'map',
'{}/{}'.format(pool, image),
'--user',
service,
'--secret',
_keyfile_path(service),
]
check_call(cmd)
def filesystem_mounted(fs):
''' Determine whether a filesytems is already mounted '''
return fs in [f for f, m in mounts()]
def make_filesystem(blk_device, fstype='ext4', timeout=10):
''' Make a new filesystem on the specified block device '''
count = 0
e_noent = os.errno.ENOENT
while not os.path.exists(blk_device):
if count >= timeout:
log('ceph: gave up waiting on block device %s' % blk_device,
level=ERROR)
raise IOError(e_noent, os.strerror(e_noent), blk_device)
log('ceph: waiting for block device %s to appear' % blk_device,
level=INFO)
count += 1
time.sleep(1)
else:
log('ceph: Formatting block device %s as filesystem %s.' %
(blk_device, fstype), level=INFO)
check_call(['mkfs', '-t', fstype, blk_device])
def place_data_on_block_device(blk_device, data_src_dst):
''' Migrate data in data_src_dst to blk_device and then remount '''
# mount block device into /mnt
mount(blk_device, '/mnt')
# copy data to /mnt
copy_files(data_src_dst, '/mnt')
# umount block device
umount('/mnt')
# Grab user/group ID's from original source
_dir = os.stat(data_src_dst)
uid = _dir.st_uid
gid = _dir.st_gid
# re-mount where the data should originally be
# TODO: persist is currently a NO-OP in core.host
mount(blk_device, data_src_dst, persist=True)
# ensure original ownership of new mount.
os.chown(data_src_dst, uid, gid)
# TODO: re-use
def modprobe(module):
''' Load a kernel module and configure for auto-load on reboot '''
log('ceph: Loading kernel module', level=INFO)
cmd = ['modprobe', module]
check_call(cmd)
with open('/etc/modules', 'r+') as modules:
if module not in modules.read():
modules.write(module)
def copy_files(src, dst, symlinks=False, ignore=None):
''' Copy files from src to dst '''
for item in os.listdir(src):
s = os.path.join(src, item)
d = os.path.join(dst, item)
if os.path.isdir(s):
shutil.copytree(s, d, symlinks, ignore)
else:
shutil.copy2(s, d)
def ensure_ceph_storage(service, pool, rbd_img, sizemb, mount_point,
blk_device, fstype, system_services=[]):
"""
NOTE: This function must only be called from a single service unit for
the same rbd_img otherwise data loss will occur.
Ensures given pool and RBD image exists, is mapped to a block device,
and the device is formatted and mounted at the given mount_point.
If formatting a device for the first time, data existing at mount_point
will be migrated to the RBD device before being re-mounted.
All services listed in system_services will be stopped prior to data
migration and restarted when complete.
"""
# Ensure pool, RBD image, RBD mappings are in place.
if not pool_exists(service, pool):
log('ceph: Creating new pool {}.'.format(pool))
create_pool(service, pool)
if not rbd_exists(service, pool, rbd_img):
log('ceph: Creating RBD image ({}).'.format(rbd_img))
create_rbd_image(service, pool, rbd_img, sizemb)
if not image_mapped(rbd_img):
log('ceph: Mapping RBD Image {} as a Block Device.'.format(rbd_img))
map_block_storage(service, pool, rbd_img)
# make file system
# TODO: What happens if for whatever reason this is run again and
# the data is already in the rbd device and/or is mounted??
# When it is mounted already, it will fail to make the fs
# XXX: This is really sketchy! Need to at least add an fstab entry
# otherwise this hook will blow away existing data if its executed
# after a reboot.
if not filesystem_mounted(mount_point):
make_filesystem(blk_device, fstype)
for svc in system_services:
if service_running(svc):
log('ceph: Stopping services {} prior to migrating data.'
.format(svc))
service_stop(svc)
place_data_on_block_device(blk_device, mount_point)
for svc in system_services:
log('ceph: Starting service {} after migrating data.'
.format(svc))
service_start(svc)
def ensure_ceph_keyring(service, user=None, group=None):
'''
Ensures a ceph keyring is created for a named service
and optionally ensures user and group ownership.
Returns False if no ceph key is available in relation state.
'''
key = None
for rid in relation_ids('ceph'):
for unit in related_units(rid):
key = relation_get('key', rid=rid, unit=unit)
if key:
break
if not key:
return False
create_keyring(service=service, key=key)
keyring = _keyring_path(service)
if user and group:
check_call(['chown', '%s.%s' % (user, group), keyring])
return True
def ceph_version():
''' Retrieve the local version of ceph '''
if os.path.exists('/usr/bin/ceph'):
cmd = ['ceph', '-v']
output = check_output(cmd)
output = output.split()
if len(output) > 3:
return output[2]
else:
return None
else:
return None

62
hooks/charmhelpers/contrib/storage/linux/loopback.py Normal file
View File

@ -0,0 +1,62 @@
import os
import re
from subprocess import (
check_call,
check_output,
)
##################################################
# loopback device helpers.
##################################################
def loopback_devices():
'''
Parse through 'losetup -a' output to determine currently mapped
loopback devices. Output is expected to look like:
/dev/loop0: [0807]:961814 (/tmp/my.img)
:returns: dict: a dict mapping {loopback_dev: backing_file}
'''
loopbacks = {}
cmd = ['losetup', '-a']
devs = [d.strip().split(' ') for d in
check_output(cmd).splitlines() if d != '']
for dev, _, f in devs:
loopbacks[dev.replace(':', '')] = re.search('\((\S+)\)', f).groups()[0]
return loopbacks
def create_loopback(file_path):
'''
Create a loopback device for a given backing file.
:returns: str: Full path to new loopback device (eg, /dev/loop0)
'''
file_path = os.path.abspath(file_path)
check_call(['losetup', '--find', file_path])
for d, f in loopback_devices().iteritems():
if f == file_path:
return d
def ensure_loopback_device(path, size):
'''
Ensure a loopback device exists for a given backing file path and size.
If it a loopback device is not mapped to file, a new one will be created.
TODO: Confirm size of found loopback device.
:returns: str: Full path to the ensured loopback device (eg, /dev/loop0)
'''
for d, f in loopback_devices().iteritems():
if f == path:
return d
if not os.path.exists(path):
cmd = ['truncate', '--size', size, path]
check_call(cmd)
return create_loopback(path)

88
hooks/charmhelpers/contrib/storage/linux/lvm.py Normal file
View File

@ -0,0 +1,88 @@
from subprocess import (
CalledProcessError,
check_call,
check_output,
Popen,
PIPE,
)
##################################################
# LVM helpers.
##################################################
def deactivate_lvm_volume_group(block_device):
'''
Deactivate any volume gruop associated with an LVM physical volume.
:param block_device: str: Full path to LVM physical volume
'''
vg = list_lvm_volume_group(block_device)
if vg:
cmd = ['vgchange', '-an', vg]
check_call(cmd)
def is_lvm_physical_volume(block_device):
'''
Determine whether a block device is initialized as an LVM PV.
:param block_device: str: Full path of block device to inspect.
:returns: boolean: True if block device is a PV, False if not.
'''
try:
check_output(['pvdisplay', block_device])
return True
except CalledProcessError:
return False
def remove_lvm_physical_volume(block_device):
'''
Remove LVM PV signatures from a given block device.
:param block_device: str: Full path of block device to scrub.
'''
p = Popen(['pvremove', '-ff', block_device],
stdin=PIPE)
p.communicate(input='y\n')
def list_lvm_volume_group(block_device):
'''
List LVM volume group associated with a given block device.
Assumes block device is a valid LVM PV.
:param block_device: str: Full path of block device to inspect.
:returns: str: Name of volume group associated with block device or None
'''
vg = None
pvd = check_output(['pvdisplay', block_device]).splitlines()
for l in pvd:
if l.strip().startswith('VG Name'):
vg = ' '.join(l.split()).split(' ').pop()
return vg
def create_lvm_physical_volume(block_device):
'''
Initialize a block device as an LVM physical volume.
:param block_device: str: Full path of block device to initialize.
'''
check_call(['pvcreate', block_device])
def create_lvm_volume_group(volume_group, block_device):
'''
Create an LVM volume group backed by a given block device.
Assumes block device has already been initialized as an LVM PV.
:param volume_group: str: Name of volume group to create.
:block_device: str: Full path of PV-initialized block device.
'''
check_call(['vgcreate', volume_group, block_device])

26
hooks/charmhelpers/contrib/storage/linux/utils.py Normal file
View File

@ -0,0 +1,26 @@
from os import stat
from stat import S_ISBLK
from subprocess import (
check_call
)
def is_block_device(path):
'''
Confirm device at path is a valid block device node.
:returns: boolean: True if path is a block device, False if not.
'''
return S_ISBLK(stat(path).st_mode)
def zap_disk(block_device):
'''
Clear a block device of partition table. Relies on sgdisk, which is
installed as pat of the 'gdisk' package in Ubuntu.
:param block_device: str: Full path of block device to clean.
'''
check_call(['sgdisk', '--zap-all', '--clear',
'--mbrtogpt', block_device])

0
hooks/charmhelpers/core/__init__.py Normal file
View File

401
hooks/charmhelpers/core/hookenv.py Normal file
View File

@ -0,0 +1,401 @@
"Interactions with the Juju environment"
# Copyright 2013 Canonical Ltd.
#
# Authors:
# Charm Helpers Developers <juju@lists.ubuntu.com>
import os
import json
import yaml
import subprocess
import sys
import UserDict
from subprocess import CalledProcessError
CRITICAL = "CRITICAL"
ERROR = "ERROR"
WARNING = "WARNING"
INFO = "INFO"
DEBUG = "DEBUG"
MARKER = object()
cache = {}
def cached(func):
"""Cache return values for multiple executions of func + args
For example:
@cached
def unit_get(attribute):
pass
unit_get('test')
will cache the result of unit_get + 'test' for future calls.
"""
def wrapper(*args, **kwargs):
global cache
key = str((func, args, kwargs))
try:
return cache[key]
except KeyError:
res = func(*args, **kwargs)
cache[key] = res
return res
return wrapper
def flush(key):
"""Flushes any entries from function cache where the
key is found in the function+args """
flush_list = []
for item in cache:
if key in item:
flush_list.append(item)
for item in flush_list:
del cache[item]
def log(message, level=None):
"""Write a message to the juju log"""
command = ['juju-log']
if level:
command += ['-l', level]
command += [message]
subprocess.call(command)
class Serializable(UserDict.IterableUserDict):
"""Wrapper, an object that can be serialized to yaml or json"""
def __init__(self, obj):
# wrap the object
UserDict.IterableUserDict.__init__(self)
self.data = obj
def __getattr__(self, attr):
# See if this object has attribute.
if attr in ("json", "yaml", "data"):
return self.__dict__[attr]
# Check for attribute in wrapped object.
got = getattr(self.data, attr, MARKER)
if got is not MARKER:
return got
# Proxy to the wrapped object via dict interface.
try:
return self.data[attr]
except KeyError:
raise AttributeError(attr)
def __getstate__(self):
# Pickle as a standard dictionary.
return self.data
def __setstate__(self, state):
# Unpickle into our wrapper.
self.data = state
def json(self):
"""Serialize the object to json"""
return json.dumps(self.data)
def yaml(self):
"""Serialize the object to yaml"""
return yaml.dump(self.data)
def execution_environment():
"""A convenient bundling of the current execution context"""
context = {}
context['conf'] = config()
if relation_id():
context['reltype'] = relation_type()
context['relid'] = relation_id()
context['rel'] = relation_get()
context['unit'] = local_unit()
context['rels'] = relations()
context['env'] = os.environ
return context
def in_relation_hook():
"""Determine whether we're running in a relation hook"""
return 'JUJU_RELATION' in os.environ
def relation_type():
"""The scope for the current relation hook"""
return os.environ.get('JUJU_RELATION', None)
def relation_id():
"""The relation ID for the current relation hook"""
return os.environ.get('JUJU_RELATION_ID', None)
def local_unit():
"""Local unit ID"""
return os.environ['JUJU_UNIT_NAME']
def remote_unit():
"""The remote unit for the current relation hook"""
return os.environ['JUJU_REMOTE_UNIT']
def service_name():
"""The name service group this unit belongs to"""
return local_unit().split('/')[0]
def hook_name():
"""The name of the currently executing hook"""
return os.path.basename(sys.argv[0])
@cached
def config(scope=None):
"""Juju charm configuration"""
config_cmd_line = ['config-get']
if scope is not None:
config_cmd_line.append(scope)
config_cmd_line.append('--format=json')
try:
return json.loads(subprocess.check_output(config_cmd_line))
except ValueError:
return None
@cached
def relation_get(attribute=None, unit=None, rid=None):
"""Get relation information"""
_args = ['relation-get', '--format=json']
if rid:
_args.append('-r')
_args.append(rid)
_args.append(attribute or '-')
if unit:
_args.append(unit)
try:
return json.loads(subprocess.check_output(_args))
except ValueError:
return None
except CalledProcessError, e:
if e.returncode == 2:
return None
raise
def relation_set(relation_id=None, relation_settings={}, **kwargs):
"""Set relation information for the current unit"""
relation_cmd_line = ['relation-set']
if relation_id is not None:
relation_cmd_line.extend(('-r', relation_id))
for k, v in (relation_settings.items() + kwargs.items()):
if v is None:
relation_cmd_line.append('{}='.format(k))
else:
relation_cmd_line.append('{}={}'.format(k, v))
subprocess.check_call(relation_cmd_line)
# Flush cache of any relation-gets for local unit
flush(local_unit())
@cached
def relation_ids(reltype=None):
"""A list of relation_ids"""
reltype = reltype or relation_type()
relid_cmd_line = ['relation-ids', '--format=json']
if reltype is not None:
relid_cmd_line.append(reltype)
return json.loads(subprocess.check_output(relid_cmd_line)) or []
return []
@cached
def related_units(relid=None):
"""A list of related units"""
relid = relid or relation_id()
units_cmd_line = ['relation-list', '--format=json']
if relid is not None:
units_cmd_line.extend(('-r', relid))
return json.loads(subprocess.check_output(units_cmd_line)) or []
@cached
def relation_for_unit(unit=None, rid=None):
"""Get the json represenation of a unit's relation"""
unit = unit or remote_unit()
relation = relation_get(unit=unit, rid=rid)
for key in relation:
if key.endswith('-list'):
relation[key] = relation[key].split()
relation['__unit__'] = unit
return relation
@cached
def relations_for_id(relid=None):
"""Get relations of a specific relation ID"""
relation_data = []
relid = relid or relation_ids()
for unit in related_units(relid):
unit_data = relation_for_unit(unit, relid)
unit_data['__relid__'] = relid
relation_data.append(unit_data)
return relation_data
@cached
def relations_of_type(reltype=None):
"""Get relations of a specific type"""
relation_data = []
reltype = reltype or relation_type()
for relid in relation_ids(reltype):
for relation in relations_for_id(relid):
relation['__relid__'] = relid
relation_data.append(relation)
return relation_data
@cached
def relation_types():
"""Get a list of relation types supported by this charm"""
charmdir = os.environ.get('CHARM_DIR', '')
mdf = open(os.path.join(charmdir, 'metadata.yaml'))
md = yaml.safe_load(mdf)
rel_types = []
for key in ('provides', 'requires', 'peers'):
section = md.get(key)
if section:
rel_types.extend(section.keys())
mdf.close()
return rel_types
@cached
def relations():
"""Get a nested dictionary of relation data for all related units"""
rels = {}
for reltype in relation_types():
relids = {}
for relid in relation_ids(reltype):
units = {local_unit(): relation_get(unit=local_unit(), rid=relid)}
for unit in related_units(relid):
reldata = relation_get(unit=unit, rid=relid)
units[unit] = reldata
relids[relid] = units
rels[reltype] = relids
return rels
@cached
def is_relation_made(relation, keys='private-address'):
'''
Determine whether a relation is established by checking for
presence of key(s). If a list of keys is provided, they
must all be present for the relation to be identified as made
'''
if isinstance(keys, str):
keys = [keys]
for r_id in relation_ids(relation):
for unit in related_units(r_id):
context = {}
for k in keys:
context[k] = relation_get(k, rid=r_id,
unit=unit)
if None not in context.values():
return True
return False
def open_port(port, protocol="TCP"):
"""Open a service network port"""
_args = ['open-port']
_args.append('{}/{}'.format(port, protocol))
subprocess.check_call(_args)
def close_port(port, protocol="TCP"):
"""Close a service network port"""
_args = ['close-port']
_args.append('{}/{}'.format(port, protocol))
subprocess.check_call(_args)
@cached
def unit_get(attribute):
"""Get the unit ID for the remote unit"""
_args = ['unit-get', '--format=json', attribute]
try:
return json.loads(subprocess.check_output(_args))
except ValueError:
return None
def unit_private_ip():
"""Get this unit's private IP address"""
return unit_get('private-address')
class UnregisteredHookError(Exception):
"""Raised when an undefined hook is called"""
pass
class Hooks(object):
"""A convenient handler for hook functions.
Example:
hooks = Hooks()
# register a hook, taking its name from the function name
@hooks.hook()
def install():
...
# register a hook, providing a custom hook name
@hooks.hook("config-changed")
def config_changed():
...
if __name__ == "__main__":
# execute a hook based on the name the program is called by
hooks.execute(sys.argv)
"""
def __init__(self):
super(Hooks, self).__init__()
self._hooks = {}
def register(self, name, function):
"""Register a hook"""
self._hooks[name] = function
def execute(self, args):
"""Execute a registered hook based on args[0]"""
hook_name = os.path.basename(args[0])
if hook_name in self._hooks:
self._hooks[hook_name]()
else:
raise UnregisteredHookError(hook_name)
def hook(self, *hook_names):
"""Decorator, registering them as hooks"""
def wrapper(decorated):
for hook_name in hook_names:
self.register(hook_name, decorated)
else:
self.register(decorated.__name__, decorated)
if '_' in decorated.__name__:
self.register(
decorated.__name__.replace('_', '-'), decorated)
return decorated
return wrapper
def charm_dir():
"""Return the root directory of the current charm"""
return os.environ.get('CHARM_DIR')

297
hooks/charmhelpers/core/host.py Normal file
View File

@ -0,0 +1,297 @@
"""Tools for working with the host system"""
# Copyright 2012 Canonical Ltd.
#
# Authors:
# Nick Moffitt <nick.moffitt@canonical.com>
# Matthew Wedgwood <matthew.wedgwood@canonical.com>
import os
import pwd
import grp
import random
import string
import subprocess
import hashlib
from collections import OrderedDict
from hookenv import log
def service_start(service_name):
"""Start a system service"""
return service('start', service_name)
def service_stop(service_name):
"""Stop a system service"""
return service('stop', service_name)
def service_restart(service_name):
"""Restart a system service"""
return service('restart', service_name)
def service_reload(service_name, restart_on_failure=False):
"""Reload a system service, optionally falling back to restart if reload fails"""
service_result = service('reload', service_name)
if not service_result and restart_on_failure:
service_result = service('restart', service_name)
return service_result
def service(action, service_name):
"""Control a system service"""
cmd = ['service', service_name, action]
return subprocess.call(cmd) == 0
def service_running(service):
"""Determine whether a system service is running"""
try:
output = subprocess.check_output(['service', service, 'status'])
except subprocess.CalledProcessError:
return False
else:
if ("start/running" in output or "is running" in output):
return True
else:
return False
def adduser(username, password=None, shell='/bin/bash', system_user=False):
"""Add a user to the system"""
try:
user_info = pwd.getpwnam(username)
log('user {0} already exists!'.format(username))
except KeyError:
log('creating user {0}'.format(username))
cmd = ['useradd']
if system_user or password is None:
cmd.append('--system')
else:
cmd.extend([
'--create-home',
'--shell', shell,
'--password', password,
])
cmd.append(username)
subprocess.check_call(cmd)
user_info = pwd.getpwnam(username)
return user_info
def add_user_to_group(username, group):
"""Add a user to a group"""
cmd = [
'gpasswd', '-a',
username,
group
]
log("Adding user {} to group {}".format(username, group))
subprocess.check_call(cmd)
def rsync(from_path, to_path, flags='-r', options=None):
"""Replicate the contents of a path"""
options = options or ['--delete', '--executability']
cmd = ['/usr/bin/rsync', flags]
cmd.extend(options)
cmd.append(from_path)
cmd.append(to_path)
log(" ".join(cmd))
return subprocess.check_output(cmd).strip()
def symlink(source, destination):
"""Create a symbolic link"""
log("Symlinking {} as {}".format(source, destination))
cmd = [
'ln',
'-sf',
source,
destination,
]
subprocess.check_call(cmd)
def mkdir(path, owner='root', group='root', perms=0555, force=False):
"""Create a directory"""
log("Making dir {} {}:{} {:o}".format(path, owner, group,
perms))
uid = pwd.getpwnam(owner).pw_uid
gid = grp.getgrnam(group).gr_gid
realpath = os.path.abspath(path)
if os.path.exists(realpath):
if force and not os.path.isdir(realpath):
log("Removing non-directory file {} prior to mkdir()".format(path))
os.unlink(realpath)
else:
os.makedirs(realpath, perms)
os.chown(realpath, uid, gid)
def write_file(path, content, owner='root', group='root', perms=0444):
"""Create or overwrite a file with the contents of a string"""
log("Writing file {} {}:{} {:o}".format(path, owner, group, perms))
uid = pwd.getpwnam(owner).pw_uid
gid = grp.getgrnam(group).gr_gid
with open(path, 'w') as target:
os.fchown(target.fileno(), uid, gid)
os.fchmod(target.fileno(), perms)
target.write(content)
def mount(device, mountpoint, options=None, persist=False):
"""Mount a filesystem at a particular mountpoint"""
cmd_args = ['mount']
if options is not None:
cmd_args.extend(['-o', options])
cmd_args.extend([device, mountpoint])
try:
subprocess.check_output(cmd_args)
except subprocess.CalledProcessError, e:
log('Error mounting {} at {}\n{}'.format(device, mountpoint, e.output))
return False
if persist:
# TODO: update fstab
pass
return True
def umount(mountpoint, persist=False):
"""Unmount a filesystem"""
cmd_args = ['umount', mountpoint]
try:
subprocess.check_output(cmd_args)
except subprocess.CalledProcessError, e:
log('Error unmounting {}\n{}'.format(mountpoint, e.output))
return False
if persist:
# TODO: update fstab
pass
return True
def mounts():
"""Get a list of all mounted volumes as [[mountpoint,device],[...]]"""
with open('/proc/mounts') as f:
# [['/mount/point','/dev/path'],[...]]
system_mounts = [m[1::-1] for m in [l.strip().split()
for l in f.readlines()]]
return system_mounts
def file_hash(path):
"""Generate a md5 hash of the contents of 'path' or None if not found """
if os.path.exists(path):
h = hashlib.md5()
with open(path, 'r') as source:
h.update(source.read()) # IGNORE:E1101 - it does have update
return h.hexdigest()
else:
return None
def restart_on_change(restart_map, stopstart=False):
"""Restart services based on configuration files changing
This function is used a decorator, for example
@restart_on_change({
'/etc/ceph/ceph.conf': [ 'cinder-api', 'cinder-volume' ]
})
def ceph_client_changed():
...
In this example, the cinder-api and cinder-volume services
would be restarted if /etc/ceph/ceph.conf is changed by the
ceph_client_changed function.
"""
def wrap(f):
def wrapped_f(*args):
checksums = {}
for path in restart_map:
checksums[path] = file_hash(path)
f(*args)
restarts = []
for path in restart_map:
if checksums[path] != file_hash(path):
restarts += restart_map[path]
services_list = list(OrderedDict.fromkeys(restarts))
if not stopstart:
for service_name in services_list:
service('restart', service_name)
else:
for action in ['stop', 'start']:
for service_name in services_list:
service(action, service_name)
return wrapped_f
return wrap
def lsb_release():
"""Return /etc/lsb-release in a dict"""
d = {}
with open('/etc/lsb-release', 'r') as lsb:
for l in lsb:
k, v = l.split('=')
d[k.strip()] = v.strip()
return d
def pwgen(length=None):
"""Generate a random pasword."""
if length is None:
length = random.choice(range(35, 45))
alphanumeric_chars = [
l for l in (string.letters + string.digits)
if l not in 'l0QD1vAEIOUaeiou']
random_chars = [
random.choice(alphanumeric_chars) for _ in range(length)]
return(''.join(random_chars))
def list_nics(nic_type):
'''Return a list of nics of given type(s)'''
if isinstance(nic_type, basestring):
int_types = [nic_type]
else:
int_types = nic_type
interfaces = []
for int_type in int_types:
cmd = ['ip', 'addr', 'show', 'label', int_type + '*']
ip_output = subprocess.check_output(cmd).split('\n')
ip_output = (line for line in ip_output if line)
for line in ip_output:
if line.split()[1].startswith(int_type):
interfaces.append(line.split()[1].replace(":", ""))
return interfaces
def set_nic_mtu(nic, mtu):
'''Set MTU on a network interface'''
cmd = ['ip', 'link', 'set', nic, 'mtu', mtu]
subprocess.check_call(cmd)
def get_nic_mtu(nic):
cmd = ['ip', 'addr', 'show', nic]
ip_output = subprocess.check_output(cmd).split('\n')
mtu = ""
for line in ip_output:
words = line.split()
if 'mtu' in words:
mtu = words[words.index("mtu") + 1]
return mtu
def get_nic_hwaddr(nic):
cmd = ['ip', '-o', '-0', 'addr', 'show', nic]
ip_output = subprocess.check_output(cmd)
hwaddr = ""
words = ip_output.split()
if 'link/ether' in words:
hwaddr = words[words.index('link/ether') + 1]
return hwaddr

5
hooks/hacluster.py
View File

@ -12,13 +12,12 @@ import subprocess
import socket
import fcntl
import struct
import lib.utils as utils
from charmhelpers.fetch import apt_install
try:
from netaddr import IPNetwork
except ImportError:
utils.install('python-netaddr')
apt_install('python-netaddr', fatal=True)
from netaddr import IPNetwork

329
hooks/hooks.py
View File

@ -14,44 +14,68 @@ import os
from base64 import b64decode
import maas as MAAS
import lib.utils as utils
import lib.cluster_utils as cluster
import pcmk
import hacluster
from charmhelpers.core.hookenv import (
log,
relation_get,
related_units,
relation_ids,
relation_set,
unit_get,
config,
Hooks, UnregisteredHookError
)
from charmhelpers.core.host import (
service_stop,
service_start,
service_restart,
service_running,
)
from charmhelpers.fetch import (
apt_install,
)
from charmhelpers.contrib.hahelpers.cluster import (
peer_units,
oldest_peer
)
hooks = Hooks()
@hooks.hook()
def install():
utils.juju_log('INFO', 'Begin install hook.')
utils.configure_source()
utils.install('corosync', 'pacemaker', 'python-netaddr', 'ipmitool')
apt_install('corosync', 'pacemaker', 'python-netaddr', 'ipmitool')
# XXX rbd OCF only included with newer versions of ceph-resource-agents.
# Bundle /w charm until we figure out a better way to install it.
if not os.path.exists('/usr/lib/ocf/resource.d/ceph'):
os.makedirs('/usr/lib/ocf/resource.d/ceph')
if not os.path.isfile('/usr/lib/ocf/resource.d/ceph/rbd'):
shutil.copy('ocf/ceph/rbd', '/usr/lib/ocf/resource.d/ceph/rbd')
utils.juju_log('INFO', 'End install hook.')
def get_corosync_conf():
conf = {}
for relid in utils.relation_ids('ha'):
for unit in utils.relation_list(relid):
for relid in relation_ids('ha'):
for unit in related_units(relid):
conf = {
'corosync_bindnetaddr':
hacluster.get_network_address(
utils.relation_get('corosync_bindiface',
unit, relid)
relation_get('corosync_bindiface',
unit, relid)
),
'corosync_mcastport': utils.relation_get('corosync_mcastport',
unit, relid),
'corosync_mcastaddr': utils.config_get('corosync_mcastaddr'),
}
'corosync_mcastport': relation_get('corosync_mcastport',
unit, relid),
'corosync_mcastaddr': config('corosync_mcastaddr'),
}
if None not in conf.itervalues():
return conf
missing = [k for k, v in conf.iteritems() if v is None]
utils.juju_log('INFO',
'Missing required principle configuration: %s' % missing)
log('Missing required principle configuration: %s' % missing)
return None
@ -60,17 +84,17 @@ def emit_corosync_conf():
corosync_conf_context = get_corosync_conf()
# write config file (/etc/corosync/corosync.conf
with open('/etc/corosync/corosync.conf', 'w') as corosync_conf:
corosync_conf.write(utils.render_template('corosync.conf',
corosync_conf_context))
corosync_conf.write(render_template('corosync.conf',
corosync_conf_context))
def emit_base_conf():
corosync_default_context = {'corosync_enabled': 'yes'}
# write /etc/default/corosync file
with open('/etc/default/corosync', 'w') as corosync_default:
corosync_default.write(utils.render_template('corosync',
corosync_default_context))
corosync_key = utils.config_get('corosync_key')
corosync_default.write(render_template('corosync',
corosync_default_context))
corosync_key = config('corosync_key')
if corosync_key:
# write the authkey
with open('/etc/corosync/authkey', 'w') as corosync_key_file:
@ -78,13 +102,12 @@ def emit_base_conf():
os.chmod = ('/etc/corosync/authkey', 0o400)
@hooks.hook()
def config_changed():
utils.juju_log('INFO', 'Begin config-changed hook.')
corosync_key = utils.config_get('corosync_key')
corosync_key = config('corosync_key')
if not corosync_key:
utils.juju_log('CRITICAL',
'No Corosync key supplied, cannot proceed')
log('CRITICAL',
'No Corosync key supplied, cannot proceed')
sys.exit(1)
hacluster.enable_lsb_services('pacemaker')
@ -98,119 +121,115 @@ def config_changed():
# Setup fencing.
configure_stonith()
utils.juju_log('INFO', 'End config-changed hook.')
@hooks.hook()
def upgrade_charm():
utils.juju_log('INFO', 'Begin upgrade-charm hook.')
install()
config_changed()
utils.juju_log('INFO', 'End upgrade-charm hook.')
def restart_corosync():
if utils.running("pacemaker"):
utils.stop("pacemaker")
utils.restart("corosync")
if service_running("pacemaker"):
service_stop("pacemaker")
service_restart("corosync")
time.sleep(5)
utils.start("pacemaker")
service_start("pacemaker")
HAMARKER = '/var/lib/juju/haconfigured'
@hooks.hook('ha-relation-joined',
'ha-relation-changed',
'hanode-relation-joined',
'hanode-relation-changed')
def configure_cluster():
# Check that we are not already configured
if os.path.exists(HAMARKER):
utils.juju_log('INFO',
'HA already configured, not reconfiguring')
log('HA already configured, not reconfiguring')
return
# Check that we are related to a principle and that
# it has already provided the required corosync configuration
if not get_corosync_conf():
utils.juju_log('WARNING',
'Unable to configure corosync right now, bailing')
log('Unable to configure corosync right now, bailing')
return
else:
utils.juju_log('INFO',
'Ready to form cluster - informing peers')
utils.relation_set(ready=True,
rid=utils.relation_ids('hanode')[0])
log('Ready to form cluster - informing peers')
relation_set(ready=True,
rid=relation_ids('hanode')[0])
# Check that there's enough nodes in order to perform the
# configuration of the HA cluster
if (len(get_cluster_nodes()) <
int(utils.config_get('cluster_count'))):
utils.juju_log('WARNING', 'Not enough nodes in cluster, bailing')
int(config('cluster_count'))):
log('Not enough nodes in cluster, bailing')
return
relids = utils.relation_ids('ha')
relids = relation_ids('ha')
if len(relids) == 1: # Should only ever be one of these
# Obtain relation information
relid = relids[0]
unit = utils.relation_list(relid)[0]
utils.juju_log('INFO',
'Using rid {} unit {}'.format(relid, unit))
unit = related_units(relid)[0]
log('Using rid {} unit {}'.format(relid, unit))
import ast
resources = \
{} if utils.relation_get("resources",
unit, relid) is None \
else ast.literal_eval(utils.relation_get("resources",
unit, relid))
{} if relation_get("resources",
unit, relid) is None \
else ast.literal_eval(relation_get("resources",
unit, relid))
resource_params = \
{} if utils.relation_get("resource_params",
unit, relid) is None \
else ast.literal_eval(utils.relation_get("resource_params",
unit, relid))
{} if relation_get("resource_params",
unit, relid) is None \
else ast.literal_eval(relation_get("resource_params",
unit, relid))
groups = \
{} if utils.relation_get("groups",
unit, relid) is None \
else ast.literal_eval(utils.relation_get("groups",
unit, relid))
{} if relation_get("groups",
unit, relid) is None \
else ast.literal_eval(relation_get("groups",
unit, relid))
ms = \
{} if utils.relation_get("ms",
unit, relid) is None \
else ast.literal_eval(utils.relation_get("ms",
unit, relid))
{} if relation_get("ms",
unit, relid) is None \
else ast.literal_eval(relation_get("ms",
unit, relid))
orders = \
{} if utils.relation_get("orders",
unit, relid) is None \
else ast.literal_eval(utils.relation_get("orders",
unit, relid))
{} if relation_get("orders",
unit, relid) is None \
else ast.literal_eval(relation_get("orders",
unit, relid))
colocations = \
{} if utils.relation_get("colocations",
unit, relid) is None \
else ast.literal_eval(utils.relation_get("colocations",
unit, relid))
{} if relation_get("colocations",
unit, relid) is None \
else ast.literal_eval(relation_get("colocations",
unit, relid))
clones = \
{} if utils.relation_get("clones",
unit, relid) is None \
else ast.literal_eval(utils.relation_get("clones",
unit, relid))
{} if relation_get("clones",
unit, relid) is None \
else ast.literal_eval(relation_get("clones",
unit, relid))
init_services = \
{} if utils.relation_get("init_services",
unit, relid) is None \
else ast.literal_eval(utils.relation_get("init_services",
unit, relid))
{} if relation_get("init_services",
unit, relid) is None \
else ast.literal_eval(relation_get("init_services",
unit, relid))
else:
utils.juju_log('WARNING',
'Related to {} ha services'.format(len(relids)))
log('Related to {} ha services'.format(len(relids)))
return
if True in [ra.startswith('ocf:openstack')
for ra in resources.itervalues()]:
utils.install('openstack-resource-agents')
apt_install('openstack-resource-agents')
if True in [ra.startswith('ocf:ceph')
for ra in resources.itervalues()]:
utils.install('ceph-resource-agents')
apt_install('ceph-resource-agents')
utils.juju_log('INFO', 'Configuring and restarting corosync')
log('Configuring and restarting corosync')
emit_corosync_conf()
restart_corosync()
utils.juju_log('INFO', 'Waiting for PCMK to start')
log('Waiting for PCMK to start')
pcmk.wait_for_pcmk()
utils.juju_log('INFO', 'Doing global cluster configuration')
log('Doing global cluster configuration')
cmd = "crm configure property stonith-enabled=false"
pcmk.commit(cmd)
cmd = "crm configure property no-quorum-policy=ignore"
@ -220,10 +239,10 @@ def configure_cluster():
pcmk.commit(cmd)
# Configure Ping service
monitor_host = utils.config_get('monitor_host')
monitor_host = config('monitor_host')
if monitor_host:
if not pcmk.crm_opt_exists('ping'):
monitor_interval = utils.config_get('monitor_interval')
monitor_interval = config('monitor_interval')
cmd = 'crm -w -F configure primitive ping' \
' ocf:pacemaker:ping params host_list="%s"' \
' multiplier="100" op monitor interval="%s"' %\
@ -235,21 +254,21 @@ def configure_cluster():
# Only configure the cluster resources
# from the oldest peer unit.
if cluster.oldest_peer(cluster.peer_units()):
utils.juju_log('INFO', 'Configuring Resources')
utils.juju_log('INFO', str(resources))
if oldest_peer(peer_units()):
log('Configuring Resources')
log(str(resources))
for res_name, res_type in resources.iteritems():
# disable the service we are going to put in HA
if res_type.split(':')[0] == "lsb":
hacluster.disable_lsb_services(res_type.split(':')[1])
if utils.running(res_type.split(':')[1]):
utils.stop(res_type.split(':')[1])
if service_running(res_type.split(':')[1]):
service_stop(res_type.split(':')[1])
elif (len(init_services) != 0 and
res_name in init_services and
init_services[res_name]):
hacluster.disable_upstart_services(init_services[res_name])
if utils.running(init_services[res_name]):
utils.stop(init_services[res_name])
if service_running(init_services[res_name]):
service_stop(init_services[res_name])
# Put the services in HA, if not already done so
# if not pcmk.is_resource_present(res_name):
if not pcmk.crm_opt_exists(res_name):
@ -262,55 +281,55 @@ def configure_cluster():
res_type,
resource_params[res_name])
pcmk.commit(cmd)
utils.juju_log('INFO', '%s' % cmd)
log('%s' % cmd)
if monitor_host:
cmd = 'crm -F configure location Ping-%s %s rule' \
' -inf: pingd lte 0' % (res_name, res_name)
pcmk.commit(cmd)
utils.juju_log('INFO', 'Configuring Groups')
utils.juju_log('INFO', str(groups))
log('Configuring Groups')
log(str(groups))
for grp_name, grp_params in groups.iteritems():
if not pcmk.crm_opt_exists(grp_name):
cmd = 'crm -w -F configure group %s %s' % (grp_name,
grp_params)
pcmk.commit(cmd)
utils.juju_log('INFO', '%s' % cmd)
log('%s' % cmd)
utils.juju_log('INFO', 'Configuring Master/Slave (ms)')
utils.juju_log('INFO', str(ms))
log('Configuring Master/Slave (ms)')
log(str(ms))
for ms_name, ms_params in ms.iteritems():
if not pcmk.crm_opt_exists(ms_name):
cmd = 'crm -w -F configure ms %s %s' % (ms_name, ms_params)
pcmk.commit(cmd)
utils.juju_log('INFO', '%s' % cmd)
log('%s' % cmd)
utils.juju_log('INFO', 'Configuring Orders')
utils.juju_log('INFO', str(orders))
log('Configuring Orders')
log(str(orders))
for ord_name, ord_params in orders.iteritems():
if not pcmk.crm_opt_exists(ord_name):
cmd = 'crm -w -F configure order %s %s' % (ord_name,
ord_params)
pcmk.commit(cmd)
utils.juju_log('INFO', '%s' % cmd)
log('%s' % cmd)
utils.juju_log('INFO', 'Configuring Colocations')
utils.juju_log('INFO', str(colocations))
log('Configuring Colocations')
log(str(colocations))
for col_name, col_params in colocations.iteritems():
if not pcmk.crm_opt_exists(col_name):
cmd = 'crm -w -F configure colocation %s %s' % (col_name,
col_params)
pcmk.commit(cmd)
utils.juju_log('INFO', '%s' % cmd)
log('%s' % cmd)
utils.juju_log('INFO', 'Configuring Clones')
utils.juju_log('INFO', str(clones))
log('Configuring Clones')
log(str(clones))
for cln_name, cln_params in clones.iteritems():
if not pcmk.crm_opt_exists(cln_name):
cmd = 'crm -w -F configure clone %s %s' % (cln_name,
cln_params)
pcmk.commit(cmd)
utils.juju_log('INFO', '%s' % cmd)
log('%s' % cmd)
for res_name, res_type in resources.iteritems():
if len(init_services) != 0 and res_name in init_services:
@ -337,9 +356,9 @@ def configure_cluster():
cmd = 'crm resource cleanup %s' % grp_name
pcmk.commit(cmd)
for rel_id in utils.relation_ids('ha'):
utils.relation_set(rid=rel_id,
clustered="yes")
for rel_id in relation_ids('ha'):
relation_set(rid=rel_id,
clustered="yes")
with open(HAMARKER, 'w') as marker:
marker.write('done')
@ -348,54 +367,52 @@ def configure_cluster():
def configure_stonith():
if utils.config_get('stonith_enabled') not in ['true', 'True']:
if config('stonith_enabled') not in ['true', 'True']:
return
if not os.path.exists(HAMARKER):
utils.juju_log('INFO',
'HA not yet configured, skipping STONITH config.')
log('HA not yet configured, skipping STONITH config.')
return
utils.juju_log('INFO', 'Configuring STONITH for all nodes in cluster.')
log('Configuring STONITH for all nodes in cluster.')
# configure stontih resources for all nodes in cluster.
# note: this is totally provider dependent and requires
# access to the MAAS API endpoint, using endpoint and credentials
# set in config.
url = utils.config_get('maas_url')
creds = utils.config_get('maas_credentials')
if None in [url, creds]:
utils.juju_log('ERROR', 'maas_url and maas_credentials must be set'
' in config to enable STONITH.')
url = config('maas_url')
creds = config('maas_credentials')
if all([url, creds]):
log('maas_url and maas_credentials must be set'
' in config to enable STONITH.')
sys.exit(1)
maas = MAAS.MAASHelper(url, creds)
nodes = maas.list_nodes()
if not nodes:
utils.juju_log('ERROR', 'Could not obtain node inventory from '
'MAAS @ %s.' % url)
log('Could not obtain node inventory from '
'MAAS @ %s.' % url)
sys.exit(1)
cluster_nodes = pcmk.list_nodes()
for node in cluster_nodes:
rsc, constraint = pcmk.maas_stonith_primitive(nodes, node)
if not rsc:
utils.juju_log('ERROR',
'Failed to determine STONITH primitive for node'
' %s' % node)
log('Failed to determine STONITH primitive for node'
' %s' % node)
sys.exit(1)
rsc_name = str(rsc).split(' ')[1]
if not pcmk.is_resource_present(rsc_name):
utils.juju_log('INFO', 'Creating new STONITH primitive %s.' %
rsc_name)
log('Creating new STONITH primitive %s.' %
rsc_name)
cmd = 'crm -F configure %s' % rsc
pcmk.commit(cmd)
if constraint:
cmd = 'crm -F configure %s' % constraint
pcmk.commit(cmd)
else:
utils.juju_log('INFO', 'STONITH primitive already exists '
'for node.')
log('STONITH primitive already exists '
'for node.')
cmd = "crm configure property stonith-enabled=true"
pcmk.commit(cmd)
@ -403,26 +420,36 @@ def configure_stonith():
def get_cluster_nodes():
hosts = []
hosts.append(utils.unit_get('private-address'))
for relid in utils.relation_ids('hanode'):
for unit in utils.relation_list(relid):
if utils.relation_get('ready',
rid=relid,
unit=unit):
hosts.append(utils.relation_get('private-address',
unit, relid))
hosts.append(unit_get('private-address'))
for relid in relation_ids('hanode'):
for unit in related_units(relid):
if relation_get('ready',
rid=relid,
unit=unit):
hosts.append(relation_get('private-address',
unit, relid))
hosts.sort()
return hosts
TEMPLATES_DIR = 'templates'
hooks = {
'install': install,
'config-changed': config_changed,
'upgrade-charm': upgrade_charm,
'ha-relation-joined': configure_cluster,
'ha-relation-changed': configure_cluster,
'hanode-relation-joined': configure_cluster,
'hanode-relation-changed': configure_cluster,
}
try:
import jinja2
except ImportError:
install('python-jinja2')
import jinja2
utils.do_hooks(hooks)
def render_template(template_name, context, template_dir=TEMPLATES_DIR):
templates = jinja2.Environment(
loader=jinja2.FileSystemLoader(template_dir)
)
template = templates.get_template(template_name)
return template.render(context)
if __name__ == '__main__':
try:
hooks.execute(sys.argv)
except UnregisteredHookError as e:
log('Unknown hook {} - skipping.'.format(e))

331
hooks/lib/utils.py
View File

@ -1,331 +0,0 @@
#
# Copyright 2012 Canonical Ltd.
#
# This file is sourced from lp:openstack-charm-helpers
#
# Authors:
# James Page <james.page@ubuntu.com>
# Paul Collins <paul.collins@canonical.com>
# Adam Gandelman <adamg@ubuntu.com>
#
import json
import os
import subprocess
import socket
import sys
def do_hooks(hooks):
hook = os.path.basename(sys.argv[0])
try:
hook_func = hooks[hook]
except KeyError:
juju_log('INFO',
"This charm doesn't know how to handle '{}'.".format(hook))
else:
hook_func()
def install(*pkgs):
cmd = [
'apt-get',
'-y',
'install'
]
for pkg in pkgs:
cmd.append(pkg)
subprocess.check_call(cmd)
TEMPLATES_DIR = 'templates'
try:
import jinja2
except ImportError:
install('python-jinja2')
import jinja2
try:
import dns.resolver
except ImportError:
install('python-dnspython')
import dns.resolver
def render_template(template_name, context, template_dir=TEMPLATES_DIR):
templates = jinja2.Environment(
loader=jinja2.FileSystemLoader(template_dir)
)
template = templates.get_template(template_name)
return template.render(context)
CLOUD_ARCHIVE = """ # Ubuntu Cloud Archive
deb http://ubuntu-cloud.archive.canonical.com/ubuntu {} main
"""
CLOUD_ARCHIVE_POCKETS = {
'folsom': 'precise-updates/folsom',
'folsom/updates': 'precise-updates/folsom',
'folsom/proposed': 'precise-proposed/folsom',
'grizzly': 'precise-updates/grizzly',
'grizzly/updates': 'precise-updates/grizzly',
'grizzly/proposed': 'precise-proposed/grizzly'
}
def configure_source():
source = str(config_get('openstack-origin'))
if not source:
return
if source.startswith('ppa:'):
cmd = [
'add-apt-repository',
source
]
subprocess.check_call(cmd)
if source.startswith('cloud:'):
# CA values should be formatted as cloud:ubuntu-openstack/pocket, eg:
# cloud:precise-folsom/updates or cloud:precise-folsom/proposed
install('ubuntu-cloud-keyring')
pocket = source.split(':')[1]
pocket = pocket.split('-')[1]
with open('/etc/apt/sources.list.d/cloud-archive.list', 'w') as apt:
apt.write(CLOUD_ARCHIVE.format(CLOUD_ARCHIVE_POCKETS[pocket]))
if source.startswith('deb'):
l = len(source.split('|'))
if l == 2:
(apt_line, key) = source.split('|')
cmd = [
'apt-key',
'adv', '--keyserver keyserver.ubuntu.com',
'--recv-keys', key
]
subprocess.check_call(cmd)
elif l == 1:
apt_line = source
with open('/etc/apt/sources.list.d/quantum.list', 'w') as apt:
apt.write(apt_line + "\n")
cmd = [
'apt-get',
'update'
]
subprocess.check_call(cmd)
# Protocols
TCP = 'TCP'
UDP = 'UDP'
def expose(port, protocol='TCP'):
cmd = [
'open-port',
'{}/{}'.format(port, protocol)
]
subprocess.check_call(cmd)
def juju_log(severity, message):
cmd = [
'juju-log',
'--log-level', severity,
message
]
subprocess.check_call(cmd)
cache = {}
def cached(func):
def wrapper(*args, **kwargs):
global cache
key = str((func, args, kwargs))
try:
return cache[key]
except KeyError:
res = func(*args, **kwargs)
cache[key] = res
return res
return wrapper
@cached
def relation_ids(relation):
cmd = [
'relation-ids',
relation
]
result = str(subprocess.check_output(cmd)).split()
if result == "":
return None
else:
return result
@cached
def relation_list(rid):
cmd = [
'relation-list',
'-r', rid,
]
result = str(subprocess.check_output(cmd)).split()
if result == "":
return None
else:
return result
@cached
def relation_get(attribute, unit=None, rid=None):
cmd = [
'relation-get',
]
if rid:
cmd.append('-r')
cmd.append(rid)
cmd.append(attribute)
if unit:
cmd.append(unit)
value = subprocess.check_output(cmd).strip() # IGNORE:E1103
if value == "":
return None
else:
return value
@cached
def relation_get_dict(relation_id=None, remote_unit=None):
"""Obtain all relation data as dict by way of JSON"""
cmd = [
'relation-get', '--format=json'
]
if relation_id:
cmd.append('-r')
cmd.append(relation_id)
if remote_unit:
remote_unit_orig = os.getenv('JUJU_REMOTE_UNIT', None)
os.environ['JUJU_REMOTE_UNIT'] = remote_unit
j = subprocess.check_output(cmd)
if remote_unit and remote_unit_orig:
os.environ['JUJU_REMOTE_UNIT'] = remote_unit_orig
d = json.loads(j)
settings = {}
# convert unicode to strings
for k, v in d.iteritems():
settings[str(k)] = str(v)
return settings
def relation_set(**kwargs):
cmd = [
'relation-set'
]
args = []
for k, v in kwargs.items():
if k == 'rid':
if v:
cmd.append('-r')
cmd.append(v)
else:
args.append('{}={}'.format(k, v))
cmd += args
subprocess.check_call(cmd)
@cached
def unit_get(attribute):
cmd = [
'unit-get',
attribute
]
value = subprocess.check_output(cmd).strip() # IGNORE:E1103
if value == "":
return None
else:
return value
@cached
def config_get(attribute):
cmd = [
'config-get',
'--format',
'json',
]
out = subprocess.check_output(cmd).strip() # IGNORE:E1103
cfg = json.loads(out)
try:
return cfg[attribute]
except KeyError:
return None
@cached
def get_unit_hostname():
return socket.gethostname()
@cached
def get_host_ip(hostname=unit_get('private-address')):
try:
# Test to see if already an IPv4 address
socket.inet_aton(hostname)
return hostname
except socket.error:
answers = dns.resolver.query(hostname, 'A')
if answers:
return answers[0].address
return None
def _svc_control(service, action):
subprocess.check_call(['service', service, action])
def restart(*services):
for service in services:
_svc_control(service, 'restart')
def stop(*services):
for service in services:
_svc_control(service, 'stop')
def start(*services):
for service in services:
_svc_control(service, 'start')
def reload(*services):
for service in services:
try:
_svc_control(service, 'reload')
except subprocess.CalledProcessError:
# Reload failed - either service does not support reload
# or it was not running - restart will fixup most things
_svc_control(service, 'restart')
def running(service):
try:
output = subprocess.check_output(['service', service, 'status'])
except subprocess.CalledProcessError:
return False
else:
if ("start/running" in output or
"is running" in output):
return True
else:
return False
def is_relation_made(relation, key='private-address'):
for r_id in (relation_ids(relation) or []):
for unit in (relation_list(r_id) or []):
if relation_get(key, rid=r_id, unit=unit):
return True
return False

9
hooks/maas.py
View File

@ -3,7 +3,8 @@ import apt_pkg as apt
import json
import subprocess
import lib.utils as utils
from charmhelpers.fetch import apt_install
from charmhelpers.core.hookenv import log, ERROR
MAAS_STABLE_PPA = 'ppa:maas-maintainers/stable '
MAAS_PROFILE_NAME = 'maas-juju-hacluster'
@ -35,14 +36,14 @@ class MAASHelper(object):
return
if not pkg.current_ver:
utils.install('maas-cli')
apt_install('maas-cli', fatal=True)
def login(self):
cmd = ['maas-cli', 'login', MAAS_PROFILE_NAME, self.url, self.creds]
try:
subprocess.check_call(cmd)
except subprocess.CalledProcessError:
utils.juju_log('ERROR', 'Could not login to MAAS @ %s.' % self.url)
log('Could not login to MAAS @ %s.' % self.url, ERROR)
return False
def logout(self):
@ -56,7 +57,7 @@ class MAASHelper(object):
cmd = ['maas-cli', MAAS_PROFILE_NAME, 'nodes', 'list']
out = subprocess.check_output(cmd)
except subprocess.CalledProcessError:
utils.juju_log('ERROR', 'Could not get node inventory from MAAS.')
log('Could not get node inventory from MAAS.', ERROR)
return False
self.logout()
return json.loads(out)

10
hooks/pcmk.py
View File

@ -1,11 +1,14 @@
import lib.utils as utils
#import lib.utils as utils
import commands
import subprocess
import socket
from charmhelpers.core.hookenv import log, ERROR
def wait_for_pcmk():
crm_up = None
hostname = utils.get_unit_hostname()
hostname = socket.gethostname()
while not crm_up:
output = commands.getstatusoutput("crm node list")[1]
crm_up = hostname in output
@ -92,8 +95,7 @@ def maas_stonith_primitive(maas_nodes, crm_node):
if power_type == 'ipmi':
rsc, constraint = _maas_ipmi_stonith_resource(crm_node, power_params)
else:
utils.juju_log('ERROR',
'Unsupported STONITH power_type: %s' % power_type)
log('Unsupported STONITH power_type: %s' % power_type, ERROR)
return False, False
if not rsc or not constraint: