~james-page/charms/trusty/rabbitmq-server/config-fixup

« back to all changes in this revision

Viewing changes to lib/charmhelpers/contrib/storage/linux/ceph.py

  • Committer: Marco Ceppi
  • Date: 2014-01-16 15:44:22 UTC
  • mfrom: (45.2.46 rabbitmq-server)
  • Revision ID: marco@ceppi.net-20140116154422-735ew4lwwdkq0b9j
[yolanda.robla] Added HA queues for Rabbit 3 as explained here: http://docs.openstack.org/high-availability-guide/content/_configure_rabbitmq.html

Show diffs side-by-side

added added

removed removed

Lines of Context:
 
1
#
 
2
# Copyright 2012 Canonical Ltd.
 
3
#
 
4
# This file is sourced from lp:openstack-charm-helpers
 
5
#
 
6
# Authors:
 
7
#  James Page <james.page@ubuntu.com>
 
8
#  Adam Gandelman <adamg@ubuntu.com>
 
9
#
 
10
 
 
11
import os
 
12
import shutil
 
13
import json
 
14
import time
 
15
 
 
16
from subprocess import (
 
17
    check_call,
 
18
    check_output,
 
19
    CalledProcessError
 
20
)
 
21
 
 
22
from charmhelpers.core.hookenv import (
 
23
    relation_get,
 
24
    relation_ids,
 
25
    related_units,
 
26
    log,
 
27
    INFO,
 
28
    WARNING,
 
29
    ERROR
 
30
)
 
31
 
 
32
from charmhelpers.core.host import (
 
33
    mount,
 
34
    mounts,
 
35
    service_start,
 
36
    service_stop,
 
37
    service_running,
 
38
    umount,
 
39
)
 
40
 
 
41
from charmhelpers.fetch import (
 
42
    apt_install,
 
43
)
 
44
 
 
45
KEYRING = '/etc/ceph/ceph.client.{}.keyring'
 
46
KEYFILE = '/etc/ceph/ceph.client.{}.key'
 
47
 
 
48
CEPH_CONF = """[global]
 
49
 auth supported = {auth}
 
50
 keyring = {keyring}
 
51
 mon host = {mon_hosts}
 
52
"""
 
53
 
 
54
 
 
55
def install():
 
56
    ''' Basic Ceph client installation '''
 
57
    ceph_dir = "/etc/ceph"
 
58
    if not os.path.exists(ceph_dir):
 
59
        os.mkdir(ceph_dir)
 
60
    apt_install('ceph-common', fatal=True)
 
61
 
 
62
 
 
63
def rbd_exists(service, pool, rbd_img):
 
64
    ''' Check to see if a RADOS block device exists '''
 
65
    try:
 
66
        out = check_output(['rbd', 'list', '--id', service,
 
67
                            '--pool', pool])
 
68
    except CalledProcessError:
 
69
        return False
 
70
    else:
 
71
        return rbd_img in out
 
72
 
 
73
 
 
74
def create_rbd_image(service, pool, image, sizemb):
 
75
    ''' Create a new RADOS block device '''
 
76
    cmd = [
 
77
        'rbd',
 
78
        'create',
 
79
        image,
 
80
        '--size',
 
81
        str(sizemb),
 
82
        '--id',
 
83
        service,
 
84
        '--pool',
 
85
        pool
 
86
    ]
 
87
    check_call(cmd)
 
88
 
 
89
 
 
90
def pool_exists(service, name):
 
91
    ''' Check to see if a RADOS pool already exists '''
 
92
    try:
 
93
        out = check_output(['rados', '--id', service, 'lspools'])
 
94
    except CalledProcessError:
 
95
        return False
 
96
    else:
 
97
        return name in out
 
98
 
 
99
 
 
100
def get_osds(service):
 
101
    '''
 
102
    Return a list of all Ceph Object Storage Daemons
 
103
    currently in the cluster
 
104
    '''
 
105
    version = ceph_version()
 
106
    if version and version >= '0.56':
 
107
        return json.loads(check_output(['ceph', '--id', service,
 
108
                                        'osd', 'ls', '--format=json']))
 
109
    else:
 
110
        return None
 
111
 
 
112
 
 
113
def create_pool(service, name, replicas=2):
 
114
    ''' Create a new RADOS pool '''
 
115
    if pool_exists(service, name):
 
116
        log("Ceph pool {} already exists, skipping creation".format(name),
 
117
            level=WARNING)
 
118
        return
 
119
    # Calculate the number of placement groups based
 
120
    # on upstream recommended best practices.
 
121
    osds = get_osds(service)
 
122
    if osds:
 
123
        pgnum = (len(osds) * 100 / replicas)
 
124
    else:
 
125
        # NOTE(james-page): Default to 200 for older ceph versions
 
126
        # which don't support OSD query from cli
 
127
        pgnum = 200
 
128
    cmd = [
 
129
        'ceph', '--id', service,
 
130
        'osd', 'pool', 'create',
 
131
        name, str(pgnum)
 
132
    ]
 
133
    check_call(cmd)
 
134
    cmd = [
 
135
        'ceph', '--id', service,
 
136
        'osd', 'pool', 'set', name,
 
137
        'size', str(replicas)
 
138
    ]
 
139
    check_call(cmd)
 
140
 
 
141
 
 
142
def delete_pool(service, name):
 
143
    ''' Delete a RADOS pool from ceph '''
 
144
    cmd = [
 
145
        'ceph', '--id', service,
 
146
        'osd', 'pool', 'delete',
 
147
        name, '--yes-i-really-really-mean-it'
 
148
    ]
 
149
    check_call(cmd)
 
150
 
 
151
 
 
152
def _keyfile_path(service):
 
153
    return KEYFILE.format(service)
 
154
 
 
155
 
 
156
def _keyring_path(service):
 
157
    return KEYRING.format(service)
 
158
 
 
159
 
 
160
def create_keyring(service, key):
 
161
    ''' Create a new Ceph keyring containing key'''
 
162
    keyring = _keyring_path(service)
 
163
    if os.path.exists(keyring):
 
164
        log('ceph: Keyring exists at %s.' % keyring, level=WARNING)
 
165
        return
 
166
    cmd = [
 
167
        'ceph-authtool',
 
168
        keyring,
 
169
        '--create-keyring',
 
170
        '--name=client.{}'.format(service),
 
171
        '--add-key={}'.format(key)
 
172
    ]
 
173
    check_call(cmd)
 
174
    log('ceph: Created new ring at %s.' % keyring, level=INFO)
 
175
 
 
176
 
 
177
def create_key_file(service, key):
 
178
    ''' Create a file containing key '''
 
179
    keyfile = _keyfile_path(service)
 
180
    if os.path.exists(keyfile):
 
181
        log('ceph: Keyfile exists at %s.' % keyfile, level=WARNING)
 
182
        return
 
183
    with open(keyfile, 'w') as fd:
 
184
        fd.write(key)
 
185
    log('ceph: Created new keyfile at %s.' % keyfile, level=INFO)
 
186
 
 
187
 
 
188
def get_ceph_nodes():
 
189
    ''' Query named relation 'ceph' to detemine current nodes '''
 
190
    hosts = []
 
191
    for r_id in relation_ids('ceph'):
 
192
        for unit in related_units(r_id):
 
193
            hosts.append(relation_get('private-address', unit=unit, rid=r_id))
 
194
    return hosts
 
195
 
 
196
 
 
197
def configure(service, key, auth):
 
198
    ''' Perform basic configuration of Ceph '''
 
199
    create_keyring(service, key)
 
200
    create_key_file(service, key)
 
201
    hosts = get_ceph_nodes()
 
202
    with open('/etc/ceph/ceph.conf', 'w') as ceph_conf:
 
203
        ceph_conf.write(CEPH_CONF.format(auth=auth,
 
204
                                         keyring=_keyring_path(service),
 
205
                                         mon_hosts=",".join(map(str, hosts))))
 
206
    modprobe('rbd')
 
207
 
 
208
 
 
209
def image_mapped(name):
 
210
    ''' Determine whether a RADOS block device is mapped locally '''
 
211
    try:
 
212
        out = check_output(['rbd', 'showmapped'])
 
213
    except CalledProcessError:
 
214
        return False
 
215
    else:
 
216
        return name in out
 
217
 
 
218
 
 
219
def map_block_storage(service, pool, image):
 
220
    ''' Map a RADOS block device for local use '''
 
221
    cmd = [
 
222
        'rbd',
 
223
        'map',
 
224
        '{}/{}'.format(pool, image),
 
225
        '--user',
 
226
        service,
 
227
        '--secret',
 
228
        _keyfile_path(service),
 
229
    ]
 
230
    check_call(cmd)
 
231
 
 
232
 
 
233
def filesystem_mounted(fs):
 
234
    ''' Determine whether a filesytems is already mounted '''
 
235
    return fs in [f for f, m in mounts()]
 
236
 
 
237
 
 
238
def make_filesystem(blk_device, fstype='ext4', timeout=10):
 
239
    ''' Make a new filesystem on the specified block device '''
 
240
    count = 0
 
241
    e_noent = os.errno.ENOENT
 
242
    while not os.path.exists(blk_device):
 
243
        if count >= timeout:
 
244
            log('ceph: gave up waiting on block device %s' % blk_device,
 
245
                level=ERROR)
 
246
            raise IOError(e_noent, os.strerror(e_noent), blk_device)
 
247
        log('ceph: waiting for block device %s to appear' % blk_device,
 
248
            level=INFO)
 
249
        count += 1
 
250
        time.sleep(1)
 
251
    else:
 
252
        log('ceph: Formatting block device %s as filesystem %s.' %
 
253
            (blk_device, fstype), level=INFO)
 
254
        check_call(['mkfs', '-t', fstype, blk_device])
 
255
 
 
256
 
 
257
def place_data_on_block_device(blk_device, data_src_dst):
 
258
    ''' Migrate data in data_src_dst to blk_device and then remount '''
 
259
    # mount block device into /mnt
 
260
    mount(blk_device, '/mnt')
 
261
    # copy data to /mnt
 
262
    copy_files(data_src_dst, '/mnt')
 
263
    # umount block device
 
264
    umount('/mnt')
 
265
    # Grab user/group ID's from original source
 
266
    _dir = os.stat(data_src_dst)
 
267
    uid = _dir.st_uid
 
268
    gid = _dir.st_gid
 
269
    # re-mount where the data should originally be
 
270
    # TODO: persist is currently a NO-OP in core.host
 
271
    mount(blk_device, data_src_dst, persist=True)
 
272
    # ensure original ownership of new mount.
 
273
    os.chown(data_src_dst, uid, gid)
 
274
 
 
275
 
 
276
# TODO: re-use
 
277
def modprobe(module):
 
278
    ''' Load a kernel module and configure for auto-load on reboot '''
 
279
    log('ceph: Loading kernel module', level=INFO)
 
280
    cmd = ['modprobe', module]
 
281
    check_call(cmd)
 
282
    with open('/etc/modules', 'r+') as modules:
 
283
        if module not in modules.read():
 
284
            modules.write(module)
 
285
 
 
286
 
 
287
def copy_files(src, dst, symlinks=False, ignore=None):
 
288
    ''' Copy files from src to dst '''
 
289
    for item in os.listdir(src):
 
290
        s = os.path.join(src, item)
 
291
        d = os.path.join(dst, item)
 
292
        if os.path.isdir(s):
 
293
            shutil.copytree(s, d, symlinks, ignore)
 
294
        else:
 
295
            shutil.copy2(s, d)
 
296
 
 
297
 
 
298
def ensure_ceph_storage(service, pool, rbd_img, sizemb, mount_point,
 
299
                        blk_device, fstype, system_services=[]):
 
300
    """
 
301
    NOTE: This function must only be called from a single service unit for
 
302
          the same rbd_img otherwise data loss will occur.
 
303
 
 
304
    Ensures given pool and RBD image exists, is mapped to a block device,
 
305
    and the device is formatted and mounted at the given mount_point.
 
306
 
 
307
    If formatting a device for the first time, data existing at mount_point
 
308
    will be migrated to the RBD device before being re-mounted.
 
309
 
 
310
    All services listed in system_services will be stopped prior to data
 
311
    migration and restarted when complete.
 
312
    """
 
313
    # Ensure pool, RBD image, RBD mappings are in place.
 
314
    if not pool_exists(service, pool):
 
315
        log('ceph: Creating new pool {}.'.format(pool))
 
316
        create_pool(service, pool)
 
317
 
 
318
    if not rbd_exists(service, pool, rbd_img):
 
319
        log('ceph: Creating RBD image ({}).'.format(rbd_img))
 
320
        create_rbd_image(service, pool, rbd_img, sizemb)
 
321
 
 
322
    if not image_mapped(rbd_img):
 
323
        log('ceph: Mapping RBD Image {} as a Block Device.'.format(rbd_img))
 
324
        map_block_storage(service, pool, rbd_img)
 
325
 
 
326
    # make file system
 
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
 
332
    #      after a reboot.
 
333
    if not filesystem_mounted(mount_point):
 
334
        make_filesystem(blk_device, fstype)
 
335
 
 
336
        for svc in system_services:
 
337
            if service_running(svc):
 
338
                log('ceph: Stopping services {} prior to migrating data.'
 
339
                    .format(svc))
 
340
                service_stop(svc)
 
341
 
 
342
        place_data_on_block_device(blk_device, mount_point)
 
343
 
 
344
        for svc in system_services:
 
345
            log('ceph: Starting service {} after migrating data.'
 
346
                .format(svc))
 
347
            service_start(svc)
 
348
 
 
349
 
 
350
def ensure_ceph_keyring(service, user=None, group=None):
 
351
    '''
 
352
    Ensures a ceph keyring is created for a named service
 
353
    and optionally ensures user and group ownership.
 
354
 
 
355
    Returns False if no ceph key is available in relation state.
 
356
    '''
 
357
    key = None
 
358
    for rid in relation_ids('ceph'):
 
359
        for unit in related_units(rid):
 
360
            key = relation_get('key', rid=rid, unit=unit)
 
361
            if key:
 
362
                break
 
363
    if not key:
 
364
        return False
 
365
    create_keyring(service=service, key=key)
 
366
    keyring = _keyring_path(service)
 
367
    if user and group:
 
368
        check_call(['chown', '%s.%s' % (user, group), keyring])
 
369
    return True
 
370
 
 
371
 
 
372
def ceph_version():
 
373
    ''' Retrieve the local version of ceph '''
 
374
    if os.path.exists('/usr/bin/ceph'):
 
375
        cmd = ['ceph', '-v']
 
376
        output = check_output(cmd)
 
377
        output = output.split()
 
378
        if len(output) > 3:
 
379
            return output[2]
 
380
        else:
 
381
            return None
 
382
    else:
 
383
        return None