2
# Copyright 2012 Canonical Ltd.
4
# This file is sourced from lp:openstack-charm-helpers
7
# James Page <james.page@ubuntu.com>
8
# Adam Gandelman <adamg@ubuntu.com>
16
from subprocess import (
21
from charmhelpers.core.hookenv import (
31
from charmhelpers.core.host import (
39
from charmhelpers.fetch import (
43
KEYRING = '/etc/ceph/ceph.client.{}.keyring'
44
KEYFILE = '/etc/ceph/ceph.client.{}.key'
46
CEPH_CONF = """[global]
47
auth supported = {auth}
49
mon host = {mon_hosts}
50
log to syslog = {use_syslog}
51
err to syslog = {use_syslog}
52
clog to syslog = {use_syslog}
57
"""Basic Ceph client installation."""
58
ceph_dir = "/etc/ceph"
59
if not os.path.exists(ceph_dir):
62
apt_install('ceph-common', fatal=True)
65
def rbd_exists(service, pool, rbd_img):
66
"""Check to see if a RADOS block device exists."""
68
out = check_output(['rbd', 'list', '--id',
69
service, '--pool', pool]).decode('UTF-8')
70
except CalledProcessError:
76
def create_rbd_image(service, pool, image, sizemb):
77
"""Create a new RADOS block device."""
78
cmd = ['rbd', 'create', image, '--size', str(sizemb), '--id', service,
83
def pool_exists(service, name):
84
"""Check to see if a RADOS pool already exists."""
86
out = check_output(['rados', '--id', service,
87
'lspools']).decode('UTF-8')
88
except CalledProcessError:
94
def get_osds(service):
95
"""Return a list of all Ceph Object Storage Daemons currently in the
98
version = ceph_version()
99
if version and version >= '0.56':
100
return json.loads(check_output(['ceph', '--id', service,
102
'--format=json']).decode('UTF-8'))
107
def create_pool(service, name, replicas=3):
108
"""Create a new RADOS pool."""
109
if pool_exists(service, name):
110
log("Ceph pool {} already exists, skipping creation".format(name),
114
# Calculate the number of placement groups based
115
# on upstream recommended best practices.
116
osds = get_osds(service)
118
pgnum = (len(osds) * 100 // replicas)
120
# NOTE(james-page): Default to 200 for older ceph versions
121
# which don't support OSD query from cli
124
cmd = ['ceph', '--id', service, 'osd', 'pool', 'create', name, str(pgnum)]
127
cmd = ['ceph', '--id', service, 'osd', 'pool', 'set', name, 'size',
132
def delete_pool(service, name):
133
"""Delete a RADOS pool from ceph."""
134
cmd = ['ceph', '--id', service, 'osd', 'pool', 'delete', name,
135
'--yes-i-really-really-mean-it']
139
def _keyfile_path(service):
140
return KEYFILE.format(service)
143
def _keyring_path(service):
144
return KEYRING.format(service)
147
def create_keyring(service, key):
148
"""Create a new Ceph keyring containing key."""
149
keyring = _keyring_path(service)
150
if os.path.exists(keyring):
151
log('Ceph keyring exists at %s.' % keyring, level=WARNING)
154
cmd = ['ceph-authtool', keyring, '--create-keyring',
155
'--name=client.{}'.format(service), '--add-key={}'.format(key)]
157
log('Created new ceph keyring at %s.' % keyring, level=DEBUG)
160
def delete_keyring(service):
161
"""Delete an existing Ceph keyring."""
162
keyring = _keyring_path(service)
163
if not os.path.exists(keyring):
164
log('Keyring does not exist at %s' % keyring, level=WARNING)
168
log('Deleted ring at %s.' % keyring, level=INFO)
171
def create_key_file(service, key):
172
"""Create a file containing key."""
173
keyfile = _keyfile_path(service)
174
if os.path.exists(keyfile):
175
log('Keyfile exists at %s.' % keyfile, level=WARNING)
178
with open(keyfile, 'w') as fd:
181
log('Created new keyfile at %s.' % keyfile, level=INFO)
184
def get_ceph_nodes():
185
"""Query named relation 'ceph' to determine current nodes."""
187
for r_id in relation_ids('ceph'):
188
for unit in related_units(r_id):
189
hosts.append(relation_get('private-address', unit=unit, rid=r_id))
194
def configure(service, key, auth, use_syslog):
195
"""Perform basic configuration of Ceph."""
196
create_keyring(service, key)
197
create_key_file(service, key)
198
hosts = get_ceph_nodes()
199
with open('/etc/ceph/ceph.conf', 'w') as ceph_conf:
200
ceph_conf.write(CEPH_CONF.format(auth=auth,
201
keyring=_keyring_path(service),
202
mon_hosts=",".join(map(str, hosts)),
203
use_syslog=use_syslog))
207
def image_mapped(name):
208
"""Determine whether a RADOS block device is mapped locally."""
210
out = check_output(['rbd', 'showmapped']).decode('UTF-8')
211
except CalledProcessError:
217
def map_block_storage(service, pool, image):
218
"""Map a RADOS block device for local use."""
222
'{}/{}'.format(pool, image),
226
_keyfile_path(service),
231
def filesystem_mounted(fs):
232
"""Determine whether a filesytems is already mounted."""
233
return fs in [f for f, m in mounts()]
236
def make_filesystem(blk_device, fstype='ext4', timeout=10):
237
"""Make a new filesystem on the specified block device."""
239
e_noent = os.errno.ENOENT
240
while not os.path.exists(blk_device):
242
log('Gave up waiting on block device %s' % blk_device,
244
raise IOError(e_noent, os.strerror(e_noent), blk_device)
246
log('Waiting for block device %s to appear' % blk_device,
251
log('Formatting block device %s as filesystem %s.' %
252
(blk_device, fstype), level=INFO)
253
check_call(['mkfs', '-t', fstype, blk_device])
256
def place_data_on_block_device(blk_device, data_src_dst):
257
"""Migrate data in data_src_dst to blk_device and then remount."""
258
# mount block device into /mnt
259
mount(blk_device, '/mnt')
261
copy_files(data_src_dst, '/mnt')
262
# umount block device
264
# Grab user/group ID's from original source
265
_dir = os.stat(data_src_dst)
268
# re-mount where the data should originally be
269
# TODO: persist is currently a NO-OP in core.host
270
mount(blk_device, data_src_dst, persist=True)
271
# ensure original ownership of new mount.
272
os.chown(data_src_dst, uid, gid)
276
def modprobe(module):
277
"""Load a kernel module and configure for auto-load on reboot."""
278
log('Loading kernel module', level=INFO)
279
cmd = ['modprobe', module]
281
with open('/etc/modules', 'r+') as modules:
282
if module not in modules.read():
283
modules.write(module)
286
def copy_files(src, dst, symlinks=False, ignore=None):
287
"""Copy files from src to dst."""
288
for item in os.listdir(src):
289
s = os.path.join(src, item)
290
d = os.path.join(dst, item)
292
shutil.copytree(s, d, symlinks, ignore)
297
def ensure_ceph_storage(service, pool, rbd_img, sizemb, mount_point,
298
blk_device, fstype, system_services=[],
300
"""NOTE: This function must only be called from a single service unit for
301
the same rbd_img otherwise data loss will occur.
303
Ensures given pool and RBD image exists, is mapped to a block device,
304
and the device is formatted and mounted at the given mount_point.
306
If formatting a device for the first time, data existing at mount_point
307
will be migrated to the RBD device before being re-mounted.
309
All services listed in system_services will be stopped prior to data
310
migration and restarted when complete.
312
# Ensure pool, RBD image, RBD mappings are in place.
313
if not pool_exists(service, pool):
314
log('Creating new pool {}.'.format(pool), level=INFO)
315
create_pool(service, pool, replicas=replicas)
317
if not rbd_exists(service, pool, rbd_img):
318
log('Creating RBD image ({}).'.format(rbd_img), level=INFO)
319
create_rbd_image(service, pool, rbd_img, sizemb)
321
if not image_mapped(rbd_img):
322
log('Mapping RBD Image {} as a Block Device.'.format(rbd_img),
324
map_block_storage(service, pool, rbd_img)
327
# TODO: What happens if for whatever reason this is run again and
328
# the data is already in the rbd device and/or is mounted??
329
# When it is mounted already, it will fail to make the fs
330
# XXX: This is really sketchy! Need to at least add an fstab entry
331
# otherwise this hook will blow away existing data if its executed
333
if not filesystem_mounted(mount_point):
334
make_filesystem(blk_device, fstype)
336
for svc in system_services:
337
if service_running(svc):
338
log('Stopping services {} prior to migrating data.'
339
.format(svc), level=DEBUG)
342
place_data_on_block_device(blk_device, mount_point)
344
for svc in system_services:
345
log('Starting service {} after migrating data.'
346
.format(svc), level=DEBUG)
350
def ensure_ceph_keyring(service, user=None, group=None):
351
"""Ensures a ceph keyring is created for a named service and optionally
352
ensures user and group ownership.
354
Returns False if no ceph key is available in relation state.
357
for rid in relation_ids('ceph'):
358
for unit in related_units(rid):
359
key = relation_get('key', rid=rid, unit=unit)
366
create_keyring(service=service, key=key)
367
keyring = _keyring_path(service)
369
check_call(['chown', '%s.%s' % (user, group), keyring])
375
"""Retrieve the local version of ceph."""
376
if os.path.exists('/usr/bin/ceph'):
378
output = check_output(cmd).decode('US-ASCII')
379
output = output.split()
388
class CephBrokerRq(object):
389
"""Ceph broker request.
391
Multiple operations can be added to a request and sent to the Ceph broker
394
Request is json-encoded for sending over the wire.
396
The API is versioned and defaults to version 1.
398
def __init__(self, api_version=1):
399
self.api_version = api_version
402
def add_op_create_pool(self, name, replica_count=3):
403
self.ops.append({'op': 'create-pool', 'name': name,
404
'replicas': replica_count})
408
return json.dumps({'api-version': self.api_version, 'ops': self.ops})
411
class CephBrokerRsp(object):
412
"""Ceph broker response.
414
Response is json-decoded and contents provided as methods/properties.
416
The API is versioned and defaults to version 1.
418
def __init__(self, encoded_rsp):
419
self.api_version = None
420
self.rsp = json.loads(encoded_rsp)
424
return self.rsp.get('exit-code')
428
return self.rsp.get('stderr')
added
removed
hooks/charmhelpers/contrib/storage/linux/ceph.py
