Merge lp:~xfactor973/charms/trusty/ceph-osd/coordinated-upgrade into lp:~openstack-charmers-archive/charms/trusty/ceph-osd/next

[Chris Holcombe]

This patch allows the ceph osd cluster to upgrade themselves one by one. It does this by using the ceph monitor cluster as a locking mechanism. There are most likely edge cases with this method that I haven't thought of. Consider this code to be lightly tested. It worked fine on ec2.

[Chris Holcombe (xfactor973)]

Note: I put the helpers up for review on charmhelpers: https://code.launchpad.net/~xfactor973/charm-helpers/ceph-keystore/+merge/287205

[uosci-testing-bot (uosci-testing-bot)]

charm_lint_check #1557 ceph-osd-next for xfactor973 mp287376
    LINT FAIL: lint-test failed

LINT Results (max last 2 lines):
make: *** [lint] Error 1
ERROR:root:Make target returned non-zero.

Full lint test output: http://paste.ubuntu.com/15210766/
Build: http://10.245.162.36:8080/job/charm_lint_check/1557/

[uosci-testing-bot (uosci-testing-bot)]

charm_unit_test #1305 ceph-osd-next for xfactor973 mp287376
    UNIT OK: passed

Build: http://10.245.162.36:8080/job/charm_unit_test/1305/

[uosci-testing-bot (uosci-testing-bot)]

charm_amulet_test #554 ceph-osd-next for xfactor973 mp287376
    AMULET FAIL: amulet-test failed

AMULET Results (max last 2 lines):
make: *** [functional_test] Error 1
ERROR:root:Make target returned non-zero.

Full amulet test output: http://paste.ubuntu.com/15210862/
Build: http://10.245.162.36:8080/job/charm_amulet_test/554/

[uosci-testing-bot (uosci-testing-bot)]

charm_unit_test #1308 ceph-osd-next for xfactor973 mp287376
    UNIT OK: passed

Build: http://10.245.162.36:8080/job/charm_unit_test/1308/

[uosci-testing-bot (uosci-testing-bot)]

charm_lint_check #1562 ceph-osd-next for xfactor973 mp287376
    LINT FAIL: lint-test failed

LINT Results (max last 2 lines):
make: *** [lint] Error 1
ERROR:root:Make target returned non-zero.

Full lint test output: http://paste.ubuntu.com/15210966/
Build: http://10.245.162.36:8080/job/charm_lint_check/1562/

[uosci-testing-bot (uosci-testing-bot)]

charm_amulet_test #558 ceph-osd-next for xfactor973 mp287376
    AMULET FAIL: amulet-test failed

AMULET Results (max last 2 lines):
make: *** [functional_test] Error 1
ERROR:root:Make target returned non-zero.

Full amulet test output: http://paste.ubuntu.com/15211124/
Build: http://10.245.162.36:8080/job/charm_amulet_test/558/

[James Page (james-page)]

I think most of my comments on the ceph-mon proposal also apply here.

review: Needs Fixing

[lp:~xfactor973/charms/trusty/ceph-osd/coordinated-upgrade]

70. By Chris Holcombe on 2016-03-01

Add back in monitor pieces. Will separate out into another MP

Unmerged revisions

70. By Chris Holcombe on 2016-03-01

Add back in monitor pieces. Will separate out into another MP

69. By Chris Holcombe on 2016-03-01

Hash the hostname instead of the ip address. That is more portable. Works now on lxc and also on ec2

68. By Chris Holcombe on 2016-02-26

Merge upstream

67. By Chris Holcombe on 2016-02-26

It rolls!. This now upgrades and rolls the ceph osd cluster one by one

=== modified file '.bzrignore'
--- .bzrignore	2015-10-30 02:23:36 +0000
+++ .bzrignore	2016-03-01 16:57:01 +0000
@@ -3,3 +3,4 @@
 .tox
 .testrepository
 bin
+.idea
 
=== modified file 'hooks/ceph.py'
--- hooks/ceph.py	2016-01-29 07:31:13 +0000
+++ hooks/ceph.py	2016-03-01 16:57:01 +0000
@@ -1,4 +1,3 @@
-
 #
 # Copyright 2012 Canonical Ltd.
 #
@@ -6,20 +5,19 @@
 #  James Page <[e-mail]>
 #  Paul Collins <[e-mail]>
 #
-
 import json
 import subprocess
 import time
 import os
 import re
 import sys
+import errno
 from charmhelpers.core.host import (
     mkdir,
     chownr,
-    service_restart,
     cmp_pkgrevno,
-    lsb_release
-)
+    lsb_release,
+    service_restart)
 from charmhelpers.core.hookenv import (
     log,
     ERROR,
@@ -54,6 +52,137 @@
         return "root"
 
 
+class CrushLocation(object):
+    def __init__(self, name, identifier, host, rack, row, datacenter, chassis, root):
+        self.name = name
+        self.identifier = identifier
+        self.host = host
+        self.rack = rack
+        self.row = row
+        self.datacenter = datacenter
+        self.chassis = chassis
+        self.root = root
+
+
+"""
+{"nodes":[{"id":-1,"name":"default","type":"root","type_id":10,"children":[-4,-3,-2]},{"id":-2,"name":"ip-172-31-10-122","type":"host","type_id":1,"children":[0]},{"id":0,"name":"osd.0","exists":1,"type":"osd","type_id":0,"status":"up","reweight":"1.000000","crush_weight":"1.000000","depth":2},{"id":-3,"name":"ip-172-31-25-187","type":"host","type_id":1,"children":[1]},{"id":1,"name":"osd.1","exists":1,"type":"osd","type_id":0,"status":"up","reweight":"1.000000","crush_weight":"1.000000","depth":2},{"id":-4,"name":"ip-172-31-38-24","type":"host","type_id":1,"children":[2]},{"id":2,"name":"osd.2","exists":1,"type":"osd","type_id":0,"status":"up","reweight":"1.000000","crush_weight":"1.000000","depth":2}],"stray":[]}
+"""
+
+
+def get_osd_tree():
+    """
+    Returns the current osd map in JSON.
+    :return: JSON String. :raise: ValueError if the monmap fails to parse.
+      Also raises CalledProcessError if our ceph command fails
+    """
+    try:
+        tree = subprocess.check_output(
+            ['sudo', '-u', ceph_user(),
+             'ceph', 'osd', 'tree', '--format=json'])
+        try:
+            json_tree = json.loads(tree)
+            crush_list = []
+            # Make sure children are present in the json
+            if not json_tree['nodes']:
+                return None
+            child_ids = json_tree['nodes'][0]['children']
+            for child in json_tree['nodes']:
+                if child['id'] in child_ids:
+                    crush_list.append(
+                        CrushLocation(
+                            name=child.get('name'),
+                            identifier=child['id'],
+                            host=child.get('host'),
+                            rack=child.get('rack'),
+                            row=child.get('row'),
+                            datacenter=child.get('datacenter'),
+                            chassis=child.get('chassis'),
+                            root=child.get('root')
+                        )
+                    )
+            return crush_list
+        except ValueError as v:
+            log("Unable to parse ceph tree json: {}. Error: {}".format(
+                tree, v.message))
+            raise
+    except subprocess.CalledProcessError as e:
+        log("ceph osd tree command failed with message: {}".format(
+            e.message))
+        raise
+
+
+def monitor_key_delete(key):
+    """
+    Deletes a key value pair on the monitor cluster.
+    :param key: String.  The key to delete.
+    """
+    try:
+        subprocess.check_output(
+            ['sudo', '-u', ceph_user(),
+             'ceph', 'config-key', 'del', str(key)])
+    except subprocess.CalledProcessError as e:
+        log("Monitor config-key put failed with message: {}".format(
+            e.message))
+        raise
+
+
+def monitor_key_set(key, value):
+    """
+    Sets a key value pair on the monitor cluster.
+    :param key: String.  The key to set.
+    :param value: The value to set.  This will be converted to a string
+        before setting
+    """
+    try:
+        subprocess.check_output(
+            ['sudo', '-u', ceph_user(),
+             'ceph', 'config-key', 'put', str(key), str(value)])
+    except subprocess.CalledProcessError as e:
+        log("Monitor config-key put failed with message: {}".format(
+            e.message))
+        raise
+
+
+def monitor_key_get(key):
+    """
+    Gets the value of an existing key in the monitor cluster.
+    :param key: String.  The key to search for.
+    :return: Returns the value of that key or None if not found.
+    """
+    try:
+        output = subprocess.check_output(
+            ['sudo', '-u', ceph_user(),
+             'ceph', 'config-key', 'get', str(key)])
+        return output
+    except subprocess.CalledProcessError as e:
+        log("Monitor config-key get failed with message: {}".format(
+            e.message))
+        return None
+
+
+def monitor_key_exists(key):
+    """
+    Searches for the existence of a key in the monitor cluster.
+    :param key: String.  The key to search for
+    :return: Returns True if the key exists, False if not and raises an
+     exception if an unknown error occurs.
+    """
+    try:
+        subprocess.check_call(
+            ['sudo', '-u', ceph_user(),
+             'ceph', 'config-key', 'exists', str(key)])
+        # I can return true here regardless because Ceph returns
+        # ENOENT if the key wasn't found
+        return True
+    except subprocess.CalledProcessError as e:
+        if e.returncode == errno.ENOENT:
+            return False
+        else:
+            log("Unknown error from ceph config-get exists: {} {}".format(
+                e.returncode, e.message))
+            raise
+
+
 def get_version():
     '''Derive Ceph release from an installed package.'''
     import apt_pkg as apt
@@ -64,7 +193,7 @@
         pkg = cache[package]
     except:
         # the package is unknown to the current apt cache.
-        e = 'Could not determine version of package with no installation '\
+        e = 'Could not determine version of package with no installation ' \
             'candidate: %s' % package
         error_out(e)
 
@@ -165,6 +294,7 @@
         # Ignore any errors for this call
         subprocess.call(cmd)
 
+
 DISK_FORMATS = [
     'xfs',
     'ext4',
@@ -211,6 +341,7 @@
 
 
 _bootstrap_keyring = "/var/lib/ceph/bootstrap-osd/ceph.keyring"
+_upgrade_keyring = "/etc/ceph/ceph.client.admin.keyring"
 
 
 def is_bootstrapped():
@@ -236,6 +367,21 @@
         ]
         subprocess.check_call(cmd)
 
+
+def import_osd_upgrade_key(key):
+    if not os.path.exists(_upgrade_keyring):
+        cmd = [
+            "sudo",
+            "-u",
+            ceph_user(),
+            'ceph-authtool',
+            _upgrade_keyring,
+            '--create-keyring',
+            '--name=client.admin',
+            '--add-key={}'.format(key)
+        ]
+        subprocess.check_call(cmd)
+
 # OSD caps taken from ceph-create-keys
 _osd_bootstrap_caps = {
     'mon': [
@@ -402,7 +548,7 @@
 
 
 def maybe_zap_journal(journal_dev):
-    if (is_osd_disk(journal_dev)):
+    if is_osd_disk(journal_dev):
         log('Looks like {} is already an OSD data'
             ' or journal, skipping.'.format(journal_dev))
         return
@@ -445,7 +591,7 @@
         log('Path {} is not a block device - bailing'.format(dev))
         return
 
-    if (is_osd_disk(dev) and not reformat_osd):
+    if is_osd_disk(dev) and not reformat_osd:
         log('Looks like {} is already an'
             ' OSD data or journal, skipping.'.format(dev))
         return
@@ -512,7 +658,7 @@
 
 
 def get_running_osds():
-    '''Returns a list of the pids of the current running OSD daemons'''
+    """Returns a list of the pids of the current running OSD daemons"""
     cmd = ['pgrep', 'ceph-osd']
     try:
         result = subprocess.check_output(cmd)
 
=== modified file 'hooks/ceph_hooks.py'
--- hooks/ceph_hooks.py	2016-02-25 15:48:22 +0000
+++ hooks/ceph_hooks.py	2016-03-01 16:57:01 +0000
@@ -8,12 +8,17 @@
 #
 
 import glob
+import hashlib
 import os
+import random
 import shutil
+import subprocess
 import sys
 import tempfile
+import time
 
 import ceph
+from charmhelpers.core import hookenv
 from charmhelpers.core.hookenv import (
     log,
     ERROR,
@@ -39,13 +44,13 @@
     filter_installed_packages,
 )
 from charmhelpers.core.sysctl import create as create_sysctl
+from charmhelpers.core import host
 
 from utils import (
     get_host_ip,
     get_networks,
     assert_charm_supports_ipv6,
-    render_template,
-)
+    render_template)
 
 from charmhelpers.contrib.openstack.alternatives import install_alternative
 from charmhelpers.contrib.network.ip import (
@@ -57,6 +62,188 @@
 
 hooks = Hooks()
 
+# A dict of valid ceph upgrade paths.  Mapping is old -> new
+upgrade_paths = {
+    'cloud:trusty-juno': 'cloud:trusty-kilo',
+    'cloud:trusty-kilo': 'cloud:trusty-liberty',
+    'cloud:trusty-liberty': None,
+}
+
+
+def pretty_print_upgrade_paths():
+    lines = []
+    for key, value in upgrade_paths.iteritems():
+        lines.append("{} -> {}".format(key, value))
+    return lines
+
+
+def check_for_upgrade():
+    c = hookenv.config()
+    old_version = c.previous('source')
+    log('old_version: {}'.format(old_version))
+    # Strip all whitespace
+    new_version = config('source')
+    if new_version:
+        # replace all whitespace
+        new_version = new_version.replace(' ', '')
+    log('new_version: {}'.format(new_version))
+
+    if old_version in upgrade_paths:
+        if new_version == upgrade_paths[old_version]:
+            log("{} to {} is a valid upgrade path.  Proceeding.".format(
+                old_version, new_version))
+            roll_osd_cluster(new_version)
+        else:
+            # Log a helpful error message
+            log("Invalid upgrade path from {} to {}.  "
+                "Valid paths are: {}".format(old_version,
+                                             new_version,
+                                             pretty_print_upgrade_paths()))
+
+
+def lock_and_roll(my_hash):
+    start_timestamp = time.time()
+
+    ceph.monitor_key_set("{}_start".format(my_hash), start_timestamp)
+    log("Rolling")
+    # This should be quick
+    upgrade_osd()
+    log("Done")
+
+    stop_timestamp = time.time()
+    # Set a key to inform others I am finished
+    ceph.monitor_key_set("{}_done".format(my_hash), stop_timestamp)
+
+
+def get_hostname():
+    try:
+        with open('/etc/hostname', 'r') as host_file:
+            host_lines = host_file.readlines()
+            if host_lines:
+                return host_lines[0].strip()
+    except IOError:
+        raise
+
+
+# TODO: Timeout busted nodes and keep moving forward
+# Edge cases:
+# 1. Previous node dies on upgrade, can we retry?
+# 2. This assumes that the osd failure domain is not set to osd.
+#    It rolls an entire server at a time.
+def roll_osd_cluster(new_version):
+    """
+    This is tricky to get right so here's what we're going to do.
+    There's 2 possible cases: Either I'm first in line or not.
+    If I'm not first in line I'll wait a random time between 5-30 seconds
+    and test to see if the previous osd is upgraded yet.
+
+    TODO: If you're not in the same failure domain it's safe to upgrade
+     1. Examine all pools and adopt the most strict failure domain policy
+        Example: Pool 1: Failure domain = rack
+        Pool 2: Failure domain = host
+        Pool 3: Failure domain = row
+
+        outcome: Failure domain = host
+    """
+    log('roll_osd_cluster called with {}'.format(new_version))
+    my_hostname = None
+    try:
+        my_hostname = get_hostname()
+    except IOError as err:
+        log("Failed to read /etc/hostname. Error: {}".format(err.message))
+        status_set('blocked', 'failed to upgrade monitor')
+
+    my_hash = hashlib.sha224(my_hostname).hexdigest()
+    # A sorted list of hashed osd names
+    osd_hashed_dict = {}
+    osd_tree_list = ceph.get_osd_tree()
+    osd_hashed_list = sorted([hashlib.sha224(
+        i.name.encode('utf-8')).hexdigest() for i in osd_tree_list])
+    # Save a hash : name mapping so we can show the user which
+    # unit name we're waiting on
+    for i in osd_tree_list:
+        osd_hashed_dict[
+            hashlib.sha224(
+                i.name.encode('utf-8')).hexdigest()
+        ] = i.name
+    log("osd_hashed_list: {}".format(osd_hashed_list))
+    try:
+        position = osd_hashed_list.index(my_hash)
+        log("upgrade position: {}".format(position))
+        if position == 0:
+            # I'm first!  Roll
+            # First set a key to inform others I'm about to roll
+            lock_and_roll(my_hash=my_hash)
+        else:
+            # Check if the previous node has finished
+            status_set('blocked',
+                       'Waiting on {} to finish upgrading'.format(
+                           osd_hashed_dict[
+                               osd_hashed_list[position - 1]]
+                       ))
+            previous_node_finished = ceph.monitor_key_exists(
+                "{}_done".format(osd_hashed_list[position - 1]))
+
+            # Block and wait on the previous nodes to finish
+            while previous_node_finished is False:
+                log("previous is not finished. Waiting")
+                # Has this node been trying to upgrade for longer than 10 minutes?
+                # If so then move on and consider that node dead.
+
+                # NOTE: This assumes the clusters clocks are somewhat accurate
+                current_timestamp = time.time()
+                previous_node_start_time = ceph.monitor_key_get(
+                    "{}_start".format(osd_hashed_list[position - 1]))
+                if (current_timestamp - (10 * 60)) > previous_node_start_time:
+                    # Previous node is probably dead.  Lets move on
+                    if previous_node_start_time is not None:
+                        log("Previous node {} appears dead. {} > {} Moving on".format(
+                            osd_hashed_dict[osd_hashed_list[position - 1]],
+                            (current_timestamp - (10 * 60)),
+                            previous_node_start_time
+                        ))
+                        lock_and_roll(my_hash=my_hash)
+                    else:
+                        # ??
+                        pass
+                else:
+                    # I have to wait.  Sleep a random amount of time and then
+                    # check if I can lock,upgrade and roll.
+                    time.sleep(random.randrange(5, 30))
+                    previous_node_finished = ceph.monitor_key_exists(
+                        "{}_done".format(osd_hashed_list[position - 1]))
+            lock_and_roll(my_hash=my_hash)
+    except ValueError:
+        log("Unable to find ceph monitor hash in list")
+        status_set('blocked', 'failed to upgrade monitor')
+
+
+def upgrade_osd():
+    add_source(config('source'), config('key'))
+
+    current_version = ceph.get_version()
+    status_set("maintenance", "Upgrading osd")
+    log("Current ceph version is {}".format(current_version))
+    new_version = config('release-version')
+    log("Upgrading to: {}".format(new_version))
+
+    try:
+        add_source(config('source'), config('key'))
+        apt_update(fatal=True)
+    except subprocess.CalledProcessError as err:
+        log("Adding the ceph source failed with message: {}".format(
+            err.message))
+        status_set("blocked", "Upgrade to {} failed".format(new_version))
+    try:
+        host.service_stop('ceph-osd-all')
+        apt_install(packages=ceph.PACKAGES, fatal=True)
+        host.service_start('ceph-osd-all')
+        status_set("active", "")
+    except subprocess.CalledProcessError as err:
+        log("Stopping ceph and upgrading packages failed "
+            "with message: {}".format(err.message))
+        status_set("blocked", "Upgrade to {} failed".format(new_version))
+
 
 def install_upstart_scripts():
     # Only install upstart configurations for older versions
@@ -113,6 +300,7 @@
     install_alternative('ceph.conf', '/etc/ceph/ceph.conf',
                         charm_ceph_conf, 90)
 
+
 JOURNAL_ZAPPED = '/var/lib/ceph/journal_zapped'
 
 
@@ -147,6 +335,9 @@
 
 @hooks.hook('config-changed')
 def config_changed():
+    # Check if an upgrade was requested
+    check_for_upgrade()
+
     # Pre-flight checks
     if config('osd-format') not in ceph.DISK_FORMATS:
         log('Invalid OSD disk format configuration specified', level=ERROR)
@@ -160,7 +351,7 @@
         create_sysctl(sysctl_dict, '/etc/sysctl.d/50-ceph-osd-charm.conf')
 
     e_mountpoint = config('ephemeral-unmount')
-    if (e_mountpoint and ceph.filesystem_mounted(e_mountpoint)):
+    if e_mountpoint and ceph.filesystem_mounted(e_mountpoint):
         umount(e_mountpoint)
     prepare_disks_and_activate()
 
@@ -189,8 +380,9 @@
     hosts = []
     for relid in relation_ids('mon'):
         for unit in related_units(relid):
-            addr = relation_get('ceph-public-address', unit, relid) or \
-                get_host_ip(relation_get('private-address', unit, relid))
+            addr = relation_get(
+                'ceph-public-address', unit, relid) or \
+                   get_host_ip(relation_get('private-address', unit, relid))
 
             if addr:
                 hosts.append('{}:6789'.format(format_ipv6_addr(addr) or addr))
@@ -246,10 +438,12 @@
             'mon-relation-departed')
 def mon_relation():
     bootstrap_key = relation_get('osd_bootstrap_key')
+    upgrade_key = relation_get('osd_upgrade_key')
     if get_fsid() and get_auth() and bootstrap_key:
         log('mon has provided conf- scanning disks')
         emit_cephconf()
         ceph.import_osd_bootstrap_key(bootstrap_key)
+        ceph.import_osd_upgrade_key(upgrade_key)
         prepare_disks_and_activate()
     else:
         log('mon cluster has not yet provided conf')
 
=== modified file 'hooks/utils.py'
--- hooks/utils.py	2016-02-18 17:10:53 +0000
+++ hooks/utils.py	2016-03-01 16:57:01 +0000
@@ -1,4 +1,3 @@
-
 #
 # Copyright 2012 Canonical Ltd.
 #
@@ -12,8 +11,8 @@
 from charmhelpers.core.hookenv import (
     unit_get,
     cached,
-    config
-)
+    config,
+    status_set)
 from charmhelpers.fetch import (
     apt_install,
     filter_installed_packages
@@ -87,6 +86,35 @@
             return answers[0].address
 
 
+def get_public_addr():
+    return get_network_addrs('ceph-public-network')[0]
+
+
+def get_network_addrs(config_opt):
+    """Get all configured public networks addresses.
+
+    If public network(s) are provided, go through them and return the
+    addresses we have configured on any of those networks.
+    """
+    addrs = []
+    networks = config(config_opt)
+    if networks:
+        networks = networks.split()
+        addrs = [get_address_in_network(n) for n in networks]
+        addrs = [a for a in addrs if a]
+
+    if not addrs:
+        if networks:
+            msg = ("Could not find an address on any of '%s' - resolve this "
+                   "error to retry" % networks)
+            status_set('blocked', msg)
+            raise Exception(msg)
+        else:
+            return [get_host_ip()]
+
+    return addrs
+
+
 def get_networks(config_opt='ceph-public-network'):
     """Get all configured networks from provided config option.
 
 
=== modified file 'templates/ceph.conf'
--- templates/ceph.conf	2016-01-18 16:42:36 +0000
+++ templates/ceph.conf	2016-03-01 16:57:01 +0000
@@ -6,7 +6,7 @@
 auth service required = {{ auth_supported }}
 auth client required = {{ auth_supported }}
 {% endif %}
-keyring = /etc/ceph/$cluster.$name.keyring
+keyring = /etc/ceph/ceph.client.admin.keyring
 mon host = {{ mon_hosts }}
 fsid = {{ fsid }}