debian, upstart, osd: osd disk preparation and activation scripts

Signed-off-by: Tommi Virtanen <tv@inktank.com>

debian/ceph.install

@@ -7,6 +7,8 @@ usr/bin/ceph-mon
 usr/bin/ceph-mds
 usr/bin/ceph-osd
 usr/bin/ceph-debugpack
+sbin/ceph-disk-prepare usr/sbin/
+sbin/ceph-disk-activate usr/sbin/
 sbin/mkcephfs
 usr/lib/ceph/ceph_common.sh
 usr/lib/rados-classes/*

debian/rules

@@ -104,6 +104,10 @@ binary-arch: build install
 	dh_install --sourcedir=$(DESTDIR) --list-missing
 	dh_installlogrotate
 	dh_installinit --no-start
+	# dh_installinit is only set up to handle one upstart script
+	# per package, so do this ourselves
+	install -d -m0755 debian/ceph/etc/init
+	install -m0644 src/upstart/*.conf debian/ceph/etc/init
 	dh_installman
 	dh_lintian
 	dh_link

src/Makefile.am

@@ -26,7 +26,9 @@ bin_PROGRAMS =
 # like bin_PROGRAMS, but these targets are only built for debug builds
 bin_DEBUGPROGRAMS =
 sbin_PROGRAMS =
-sbin_SCRIPTS =
+sbin_SCRIPTS = \
+	ceph-disk-prepare \
+	ceph-disk-activate
 bin_SCRIPTS = ceph-run $(srcdir)/ceph-clsinfo ceph-debugpack ceph-rbdnamer
 dist_bin_SCRIPTS =
 # C/C++ tests to build will be appended to this
@@ -882,7 +884,14 @@ EXTRA_DIST += \
 	$(srcdir)/ceph-rbdnamer \
 	$(ceph_tool_gui_DATA) \
 	$(srcdir)/test/encoding/readable.sh \
-	$(srcdir)/test/encoding/check-generated.sh
+	$(srcdir)/test/encoding/check-generated.sh \
+	$(srcdir)/upstart/ceph-mon-all.conf \
+	$(srcdir)/upstart/ceph-mon.conf \
+	$(srcdir)/upstart/ceph-osd.conf \
+	$(srcdir)/upstart/ceph-hotplug.conf \
+	ceph-disk-prepare \
+	ceph-disk-activate
+
 
 EXTRA_DIST += $(srcdir)/$(shell_scripts:%=%.in)
 

src/ceph-disk-activate

@@ -0,0 +1,489 @@
+#!/usr/bin/python
+
+import argparse
+import errno
+import logging
+import os
+import os.path
+import re
+import subprocess
+import sys
+import tempfile
+
+
+log_name = __name__
+if log_name == '__main__':
+    log_name = os.path.basename(sys.argv[0])
+log = logging.getLogger(log_name)
+
+
+class ActivateError(Exception):
+    """
+    OSD activation error
+    """
+
+    def __str__(self):
+        doc = self.__doc__.strip()
+        return ': '.join([doc] + [str(a) for a in self.args])
+
+
+class BadMagicError(ActivateError):
+    """
+    Does not look like a Ceph OSD, or incompatible version
+    """
+
+
+class TruncatedLineError(ActivateError):
+    """
+    Line is truncated
+    """
+
+
+class TooManyLinesError(ActivateError):
+    """
+    Too many lines
+    """
+
+
+class FilesystemTypeError(ActivateError):
+    """
+    Cannot discover filesystem type
+    """
+
+
+class MountError(ActivateError):
+    """
+    Mounting filesystem failed
+    """
+
+
+def maybe_mkdir(*a, **kw):
+    try:
+        os.mkdir(*a, **kw)
+    except OSError, e:
+        if e.errno == errno.EEXIST:
+            pass
+        else:
+            raise
+
+
+def must_be_one_line(line):
+    if line[-1:] != '\n':
+        raise TruncatedLineError(line)
+    line = line[:-1]
+    if '\n' in line:
+        raise TooManyLinesError(line)
+    return line
+
+
+def read_one_line(parent, name):
+    """
+    Read a file whose sole contents are a single line.
+
+    Strips the newline.
+
+    :return: Contents of the line, or None if file did not exist.
+    """
+    path = os.path.join(parent, name)
+    try:
+        line = file(path, 'rb').read()
+    except IOError as e:
+        if e.errno == errno.ENOENT:
+            return None
+        else:
+            raise
+
+    try:
+        line = must_be_one_line(line)
+    except (TruncatedLineError, TooManyLinesError) as e:
+        raise ActivateError('File is corrupt: {path}: {msg}'.format(
+                path=path,
+                msg=e,
+                ))
+    return line
+
+
+def write_one_line(parent, name, text):
+    """
+    Write a file whose sole contents are a single line.
+
+    Adds a newline.
+    """
+    path = os.path.join(parent, name)
+    tmp = '{path}.{pid}.tmp'.format(path=path, pid=os.getpid())
+    with file(tmp, 'wb') as f:
+        f.write(text + '\n')
+        os.fsync(f.fileno())
+    os.rename(tmp, path)
+
+
+CEPH_OSD_ONDISK_MAGIC = 'ceph osd volume v026'
+
+
+def check_osd_magic(path):
+    """
+    Check that this path has the Ceph OSD magic.
+
+    :raises: BadMagicError if this does not look like a Ceph OSD data
+    dir.
+    """
+    magic = read_one_line(path, 'magic')
+    if magic is None:
+        # probably not mkfs'ed yet
+        raise BadMagicError(path)
+    if magic != CEPH_OSD_ONDISK_MAGIC:
+        raise BadMagicError(path)
+
+
+def check_osd_id(osd_id):
+    """
+    Ensures osd id is numeric.
+    """
+    if not re.match(r'^[0-9]+$', osd_id):
+        raise ActivateError('osd id is not numeric')
+
+
+def get_osd_id(path):
+    osd_id = read_one_line(path, 'whoami')
+    if osd_id is not None:
+        check_osd_id(osd_id)
+    return osd_id
+
+
+# TODO depend on python2.7
+def _check_output(*args, **kwargs):
+    process = subprocess.Popen(
+        stdout=subprocess.PIPE,
+        *args, **kwargs)
+    out, _ = process.communicate()
+    ret = process.wait()
+    if ret:
+        cmd = kwargs.get("args")
+        if cmd is None:
+            cmd = args[0]
+        raise subprocess.CalledProcessError(ret, cmd, output=out)
+    return out
+
+
+def allocate_osd_id(
+    cluster,
+    fsid,
+    keyring,
+    ):
+    log.debug('Allocating OSD id...')
+    try:
+        osd_id = _check_output(
+            args=[
+                'ceph',
+                '--cluster', cluster,
+                '--name', 'client.bootstrap-osd',
+                '--keyring', keyring,
+                'osd', 'create', '--concise',
+                fsid,
+                ],
+            )
+    except subprocess.CalledProcessError as e:
+        raise ActivateError('ceph osd create failed', e)
+    osd_id = must_be_one_line(osd_id)
+    check_osd_id(osd_id)
+    return osd_id
+
+
+def mkfs(
+    path,
+    cluster,
+    osd_id,
+    fsid,
+    keyring,
+    ):
+    monmap = os.path.join(path, 'activate.monmap')
+    subprocess.check_call(
+        args=[
+            'ceph',
+            '--cluster', cluster,
+            '--name', 'client.bootstrap-osd',
+            '--keyring', keyring,
+            'mon', 'getmap', '-o', monmap,
+            ],
+        )
+
+    subprocess.check_call(
+        args=[
+            'ceph-osd',
+            '--cluster', cluster,
+            '--mkfs',
+            '--mkkey',
+            '-i', osd_id,
+            '--monmap', monmap,
+            '--osd-data', path,
+            '--osd-journal', os.path.join(path, 'journal'),
+            '--osd-uuid', fsid,
+            '--keyring', os.path.join(path, 'keyring'),
+            ],
+        )
+    # TODO ceph-osd --mkfs removes the monmap file?
+    # os.unlink(monmap)
+
+
+def auth_key(
+    path,
+    cluster,
+    osd_id,
+    keyring,
+    ):
+    subprocess.check_call(
+        args=[
+            'ceph',
+            '--cluster', cluster,
+            '--name', 'client.bootstrap-osd',
+            '--keyring', keyring,
+            'auth', 'add', 'osd.{osd_id}'.format(osd_id=osd_id),
+            '-i', os.path.join(path, 'keyring'),
+            'osd', 'allow *',
+            'mon', 'allow rwx',
+            ],
+        )
+
+
+def move_mount(
+    path,
+    cluster,
+    osd_id,
+    ):
+    log.debug('Moving mount to final location...')
+    parent = '/var/lib/ceph/osd'
+    osd_data = os.path.join(
+        parent,
+        '{cluster}-{osd_id}'.format(cluster=cluster, osd_id=osd_id),
+        )
+    maybe_mkdir(osd_data)
+    subprocess.check_call(
+        args=[
+            'mount',
+            '--move',
+            '--',
+            path,
+            osd_data,
+            ],
+        )
+
+
+def upstart_start(
+    cluster,
+    osd_id,
+    ):
+    log.debug('Starting service...')
+    subprocess.check_call(
+        args=[
+            'initctl',
+            'start',
+            # since the daemon starting doesn't guarantee much about
+            # the service being operational anyway, don't bother
+            # waiting for it
+            '--no-wait',
+            '--',
+            'ceph-osd',
+            'cluster={cluster}'.format(cluster=cluster),
+            'id={osd_id}'.format(osd_id=osd_id),
+            ],
+        )
+
+
+def detect_fstype(
+    dev,
+    ):
+    fstype = _check_output(
+        args=[
+            'blkid',
+            # we don't want stale cached results
+            '-p',
+            '-s', 'TYPE',
+            '-o' 'value',
+            '--',
+            dev,
+            ],
+        )
+    fstype = must_be_one_line(fstype)
+    return fstype
+
+
+MOUNT_OPTIONS = dict(
+    ext4='user_xattr',
+    )
+
+
+def mount(
+    dev,
+    ):
+    # pick best-of-breed mount options based on fs type
+    try:
+        fstype = detect_fstype(dev)
+    except (subprocess.CalledProcessError,
+            TruncatedLineError,
+            TooManyLinesError) as e:
+        raise FilesystemTypeError(
+            'device {dev}'.format(dev=dev),
+            e,
+            )
+    options = MOUNT_OPTIONS.get(fstype, '')
+
+    # mount
+    path = tempfile.mkdtemp(
+        prefix='mnt.',
+        dir='/var/lib/ceph/tmp',
+        )
+    try:
+        subprocess.check_call(
+            args=[
+                'mount',
+                '-o', options,
+                '--',
+                dev,
+                path,
+                ],
+            )
+    except subprocess.CalledProcessError as e:
+        try:
+            os.rmdir(path)
+        except (OSError, IOError):
+            pass
+        raise MountError(e)
+
+    return path
+
+
+def activate(
+    path,
+    activate_key_template,
+    do_mount,
+    ):
+    if do_mount:
+        path = mount(dev=path)
+
+    # TODO unmount on errors?
+
+    check_osd_magic(path)
+
+    ceph_fsid = read_one_line(path, 'ceph_fsid')
+    if ceph_fsid is None:
+        raise ActivateError('No cluster uuid assigned.')
+    log.debug('Cluster uuid is %s', ceph_fsid)
+
+    # TODO use ceph_fsid to find the right cluster
+    cluster = 'ceph'
+    log.debug('Cluster name is %s', cluster)
+
+    fsid = read_one_line(path, 'fsid')
+    if fsid is None:
+        raise ActivateError('No OSD uuid assigned.')
+    log.debug('OSD uuid is %s', fsid)
+
+    keyring = activate_key_template.format(cluster=cluster)
+
+    osd_id = get_osd_id(path)
+    if osd_id is None:
+        osd_id = allocate_osd_id(
+            cluster=cluster,
+            fsid=fsid,
+            keyring=keyring,
+            )
+        write_one_line(path, 'whoami', osd_id)
+    log.debug('OSD id is %s', osd_id)
+
+    if not os.path.exists(os.path.join(path, 'ready')):
+        log.debug('Initializing OSD...')
+        # re-running mkfs is safe, so just run until it completes
+        mkfs(
+            path=path,
+            cluster=cluster,
+            osd_id=osd_id,
+            fsid=fsid,
+            keyring=keyring,
+            )
+
+    if not os.path.exists(os.path.join(path, 'active')):
+        log.debug('Authorizing OSD key...')
+        auth_key(
+            path=path,
+            cluster=cluster,
+            osd_id=osd_id,
+            keyring=keyring,
+            )
+        write_one_line(path, 'active', 'ok')
+
+    move_mount(
+        path=path,
+        cluster=cluster,
+        osd_id=osd_id,
+        )
+
+    if do_mount:
+        # if we created a temp dir to mount it, remove it
+        os.rmdir(path)
+
+    upstart_start(
+        cluster=cluster,
+        osd_id=osd_id,
+        )
+
+
+def parse_args():
+    parser = argparse.ArgumentParser(
+        description='Activate a Ceph OSD',
+        )
+    parser.add_argument(
+        '-v', '--verbose',
+        action='store_true', default=None,
+        help='be more verbose',
+        )
+    parser.add_argument(
+        '--mount',
+        action='store_true', default=None,
+        help='mount the device first',
+        )
+    parser.add_argument(
+        '--activate-key',
+        metavar='PATH',
+        help='bootstrap-osd keyring path template (%(default)s)',
+        dest='activate_key_template',
+        )
+    parser.add_argument(
+        'path',
+        metavar='PATH',
+        help='path to OSD data directory, or block device if using --mount',
+        )
+    parser.set_defaults(
+        activate_key_template='/var/lib/ceph/bootstrap-osd/{cluster}.keyring',
+        # we want to hold on to this, for later
+        prog=parser.prog,
+        )
+    args = parser.parse_args()
+    return args
+
+
+def main():
+    args = parse_args()
+
+    loglevel = logging.INFO
+    if args.verbose:
+        loglevel = logging.DEBUG
+
+    logging.basicConfig(
+        level=loglevel,
+        )
+
+    try:
+        activate(
+            path=args.path,
+            activate_key_template=args.activate_key_template,
+            do_mount=args.mount,
+            )
+    except ActivateError as e:
+        print >>sys.stderr, '{prog}: {msg}'.format(
+            prog=args.prog,
+            msg=e,
+            )
+        sys.exit(1)
+
+if __name__ == '__main__':
+    main()

src/ceph-disk-prepare

@@ -0,0 +1,115 @@
+#!/usr/bin/python
+
+import argparse
+import logging
+import os
+import os.path
+import sys
+import uuid
+
+
+log_name = __name__
+if log_name == '__main__':
+    log_name = os.path.basename(sys.argv[0])
+log = logging.getLogger(log_name)
+
+
+class PrepareError(Exception):
+    """
+    OSD preparation error
+    """
+
+    def __str__(self):
+        doc = self.__doc__.strip()
+        return ': '.join([doc] + [str(a) for a in self.args])
+
+
+def write_one_line(parent, name, text):
+    """
+    Write a file whose sole contents are a single line.
+
+    Adds a newline.
+    """
+    path = os.path.join(parent, name)
+    tmp = '{path}.{pid}.tmp'.format(path=path, pid=os.getpid())
+    with file(tmp, 'wb') as f:
+        f.write(text + '\n')
+        os.fsync(f.fileno())
+    os.rename(tmp, path)
+
+
+CEPH_OSD_ONDISK_MAGIC = 'ceph osd volume v026'
+
+
+def prepare(
+    path,
+    cluster_uuid,
+    ):
+    """
+    Prepare a disk to be used as an OSD data disk.
+
+    The ``magic`` file is written last, so it's presence is a reliable
+    indicator of the whole sequence having completed.
+    """
+    if os.path.exists(os.path.join(path, 'magic')):
+        raise PrepareError('already prepared, aborting')
+
+    write_one_line(path, 'ceph_fsid', cluster_uuid)
+    osd_uuid = str(uuid.uuid4())
+    write_one_line(path, 'fsid', osd_uuid)
+    write_one_line(path, 'magic', CEPH_OSD_ONDISK_MAGIC)
+
+
+def parse_args():
+    parser = argparse.ArgumentParser(
+        description='Prepare a disk for a Ceph OSD',
+        )
+    parser.add_argument(
+        '-v', '--verbose',
+        action='store_true', default=None,
+        help='be more verbose',
+        )
+    parser.add_argument(
+        '--cluster-uuid',
+        metavar='UUID',
+        help='cluster uuid to assign this disk to',
+        required=True,
+        )
+    parser.add_argument(
+        'path',
+        metavar='PATH',
+        help='path to OSD data directory',
+        )
+    parser.set_defaults(
+        # we want to hold on to this, for later
+        prog=parser.prog,
+        )
+    args = parser.parse_args()
+    return args
+
+
+def main():
+    args = parse_args()
+
+    loglevel = logging.INFO
+    if args.verbose:
+        loglevel = logging.DEBUG
+
+    logging.basicConfig(
+        level=loglevel,
+        )
+
+    try:
+        prepare(
+            path=args.path,
+            cluster_uuid=args.cluster_uuid,
+            )
+    except PrepareError as e:
+        print >>sys.stderr, '{prog}: {msg}'.format(
+            prog=args.prog,
+            msg=e,
+            )
+        sys.exit(1)
+
+if __name__ == '__main__':
+    main()

src/upstart/ceph-hotplug.conf

@@ -0,0 +1,10 @@
+description "Ceph hotplug"
+
+start on block-device-added \
+  DEVTYPE=partition \
+  ID_PART_ENTRY_TYPE=4fbd7e29-9d25-41b8-afd0-062c0ceff05d
+
+task
+instance $DEVNAME
+
+exec /usr/sbin/ceph-disk-activate --mount -- "$DEVNAME"

src/upstart/ceph-mon-all.conf

@@ -0,0 +1,26 @@
+description "Ceph MON (start all instances)"
+
+start on filesystem
+
+task
+
+script
+  set -e
+  # TODO what's the valid charset for cluster names and mon ids?
+  find /var/lib/ceph/mon/ -mindepth 1 -maxdepth 1 -regextype posix-egrep -regex '.*/[a-z0-9]+-[a-z0-9]+' -printf '%P\n' \
+  | while read f; do
+    if [ -e "/var/lib/ceph/mon/$f/done" ]; then
+        cluster="${f%%-*}"
+        id="${f#*-}"
+
+	# upstart start(8) fails if the job is already running
+	# https://bugs.launchpad.net/upstart/+bug/878322
+
+	# also cannot ask for status of instance that isn't running at
+	# that time, so just list all and filter that
+	if initctl list|mawk '$1=="ceph-mon" && $2=="(" CLUSTER "/" INSTANCE ")" && $3~/start\// { exit 1 }' CLUSTER="$cluster" INSTANCE="$id"; then
+	  start ceph-mon cluster="$cluster" id="$id"
+	fi
+    fi
+  done
+end script

src/upstart/ceph-mon.conf

@@ -0,0 +1,18 @@
+description "Ceph MON"
+
+stop on runlevel [!2345]
+
+respawn
+respawn limit 5 30
+
+pre-start script
+    set -e
+    test -x /usr/bin/ceph-mon || { stop; exit 0; }
+    test -d "/var/lib/ceph/mon/${cluster:-ceph}-$id" || { stop; exit 0; }
+
+    install -d -m0755 /var/run/ceph
+end script
+
+instance ${cluster:-ceph}/$id
+
+exec /usr/bin/ceph-mon --cluster="${cluster:-ceph}" -i "$id" -f

src/upstart/ceph-osd.conf

@@ -0,0 +1,28 @@
+description "Ceph OSD"
+
+stop on runlevel [!2345]
+
+respawn
+respawn limit 5 30
+
+pre-start script
+    set -e
+    test -x /usr/bin/ceph-osd || { stop; exit 0; }
+    test -d "/var/lib/ceph/osd/${cluster:-ceph}-$id" || { stop; exit 0; }
+
+    install -d -m0755 /var/run/ceph
+
+    # update location in crush
+    # TODO: un-hardcode the domain=root assumption
+    ceph \
+      --cluster="${cluster:-ceph}" \
+      --name="osd.$id" \
+      --keyring="/var/lib/ceph/osd/${cluster:-ceph}-$id/keyring" \
+      osd crush set \
+      "$id" "osd.$id" 1 domain=root \
+      || :
+end script
+
+instance ${cluster:-ceph}/$id
+
+exec /usr/bin/ceph-osd --cluster="${cluster:-ceph}" -i "$id" -f