Adding upstream version 14.2.21.upstream/14.2.21upstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

13 files changed, 2187 insertions, 0 deletions

diff --git a/qa/workunits/mon/auth_caps.sh b/qa/workunits/mon/auth_caps.sh
new file mode 100755
index 00000000..1f59ae1f
--- /dev/null
+++ b/qa/workunits/mon/auth_caps.sh
@@ -0,0 +1,130 @@
+#!/usr/bin/env bash
+
+set -e
+set -x
+declare -A keymap
+
+combinations="r w x rw rx wx rwx"
+
+for i in ${combinations}; do
+  k="foo_$i"
+  k=`ceph auth get-or-create-key client.$i mon "allow $i"` || exit 1
+  keymap["$i"]=$k
+done
+
+# add special caps
+keymap["all"]=`ceph auth get-or-create-key client.all mon 'allow *'` || exit 1
+
+tmp=`mktemp`
+ceph auth export > $tmp
+
+trap "rm $tmp" INT ERR EXIT QUIT 0
+
+expect() {
+
+  set +e
+
+  local expected_ret=$1
+  local ret
+
+  shift
+  cmd=$@
+
+  eval $cmd
+  ret=$?
+
+  set -e
+
+  if [[ $ret -ne $expected_ret ]]; then
+    echo "ERROR: running \'$cmd\': expected $expected_ret got $ret"
+    return 1
+  fi
+
+  return 0
+}
+
+read_ops() {
+  local caps=$1
+  local has_read=1 has_exec=1
+  local ret
+  local args
+
+  ( echo $caps | grep 'r' ) || has_read=0
+  ( echo $caps | grep 'x' ) || has_exec=0
+  
+  if [[ "$caps" == "all" ]]; then
+    has_read=1
+    has_exec=1
+  fi
+
+  ret=13
+  if [[ $has_read -gt 0 && $has_exec -gt 0 ]]; then
+    ret=0
+  fi
+
+  args="--id $caps --key ${keymap[$caps]}"
+ 
+  expect $ret ceph auth get client.admin $args
+  expect $ret ceph auth get-key client.admin $args
+  expect $ret ceph auth export $args
+  expect $ret ceph auth export client.admin $args
+  expect $ret ceph auth ls $args
+  expect $ret ceph auth print-key client.admin $args
+  expect $ret ceph auth print_key client.admin $args
+}
+
+write_ops() {
+
+  local caps=$1
+  local has_read=1 has_write=1 has_exec=1
+  local ret
+  local args
+
+  ( echo $caps | grep 'r' ) || has_read=0
+  ( echo $caps | grep 'w' ) || has_write=0
+  ( echo $caps | grep 'x' ) || has_exec=0
+
+  if [[ "$caps" == "all" ]]; then
+    has_read=1
+    has_write=1
+    has_exec=1
+  fi
+
+  ret=13
+  if [[ $has_read -gt 0 && $has_write -gt 0 && $has_exec -gt 0 ]]; then
+    ret=0
+  fi
+
+  args="--id $caps --key ${keymap[$caps]}"
+
+  expect $ret ceph auth add client.foo $args
+  expect $ret "ceph auth caps client.foo mon 'allow *' $args"
+  expect $ret ceph auth get-or-create client.admin $args
+  expect $ret ceph auth get-or-create-key client.admin $args
+  expect $ret ceph auth get-or-create-key client.baz $args
+  expect $ret ceph auth del client.foo $args
+  expect $ret ceph auth del client.baz $args
+  expect $ret ceph auth import -i $tmp $args
+}
+
+echo "running combinations: ${!keymap[@]}"
+
+subcmd=$1
+
+for i in ${!keymap[@]}; do
+  echo "caps: $i"
+  if [[ -z "$subcmd" || "$subcmd" == "read" || "$subcmd" == "all" ]]; then
+    read_ops $i
+  fi
+
+  if [[ -z "$subcmd" || "$subcmd" == "write" || "$subcmd" == "all" ]]; then
+    write_ops $i
+  fi
+done
+
+# cleanup
+for i in ${combinations} all; do
+  ceph auth del client.$i || exit 1
+done
+
+echo "OK"
diff --git a/qa/workunits/mon/caps.py b/qa/workunits/mon/caps.py
new file mode 100644
index 00000000..2634f776
--- /dev/null
+++ b/qa/workunits/mon/caps.py
@@ -0,0 +1,362 @@
+#!/usr/bin/python
+
+from __future__ import print_function
+
+import subprocess
+import shlex
+import errno
+import sys
+import os
+import io
+import re
+
+import six
+
+from ceph_argparse import * # noqa
+
+keyring_base = '/tmp/cephtest-caps.keyring'
+
+class UnexpectedReturn(Exception):
+  def __init__(self, cmd, ret, expected, msg):
+    if isinstance(cmd, list):
+      self.cmd = ' '.join(cmd)
+    else:
+      assert isinstance(cmd, str) or isinstance(cmd, six.text_type), \
+          'cmd needs to be either a list or a str'
+      self.cmd = cmd
+    self.cmd = str(self.cmd)
+    self.ret = int(ret)
+    self.expected = int(expected)
+    self.msg = str(msg)
+
+  def __str__(self):
+    return repr('{c}: expected return {e}, got {r} ({o})'.format(
+        c=self.cmd, e=self.expected, r=self.ret, o=self.msg))
+
+def call(cmd):
+  if isinstance(cmd, list):
+    args = cmd
+  elif isinstance(cmd, str) or isinstance(cmd, six.text_type):
+    args = shlex.split(cmd)
+  else:
+    assert False, 'cmd is not a string/unicode nor a list!'
+
+  print('call: {0}'.format(args))
+  proc = subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+  ret = proc.wait()
+
+  return (ret, proc)
+
+def expect(cmd, expected_ret):
+
+  try:
+    (r, p) = call(cmd)
+  except ValueError as e:
+    print('unable to run {c}: {err}'.format(c=repr(cmd), err=e.message),
+          file=sys.stderr)
+    return errno.EINVAL
+
+  assert r == p.returncode, \
+      'wth? r was supposed to match returncode!'
+
+  if r != expected_ret:
+    raise UnexpectedReturn(repr(cmd), r, expected_ret, str(p.stderr.read()))
+
+  return p
+
+def expect_to_file(cmd, expected_ret, out_file, mode='a'):
+
+  # Let the exception be propagated to the caller
+  p = expect(cmd, expected_ret)
+  assert p.returncode == expected_ret, \
+      'expected result doesn\'t match and no exception was thrown!'
+
+  with io.open(out_file, mode) as file:
+    file.write(six.text_type(p.stdout.read()))
+
+  return p
+
+class Command:
+  def __init__(self, cid, j):
+    self.cid = cid[3:]
+    self.perms = j['perm']
+    self.module = j['module']
+
+    self.sig = ''
+    self.args = []
+    for s in j['sig']:
+      if not isinstance(s, dict):
+        assert isinstance(s, str) or isinstance(s,six.text_type), \
+            'malformatted signature cid {0}: {1}\n{2}'.format(cid,s,j)
+        if len(self.sig) > 0:
+          self.sig += ' '
+        self.sig += s
+      else:
+        self.args.append(s)
+
+  def __str__(self):
+    return repr('command {0}: {1} (requires \'{2}\')'.format(self.cid,\
+          self.sig, self.perms))
+
+
+def destroy_keyring(path):
+  if not os.path.exists(path):
+    raise Exception('oops! cannot remove inexistent keyring {0}'.format(path))
+
+  # grab all client entities from the keyring
+  entities = [m.group(1) for m in [re.match(r'\[client\.(.*)\]', l)
+                for l in [str(line.strip())
+                  for line in io.open(path,'r')]] if m is not None]
+
+  # clean up and make sure each entity is gone
+  for e in entities:
+    expect('ceph auth del client.{0}'.format(e), 0)
+    expect('ceph auth get client.{0}'.format(e), errno.ENOENT)
+
+  # remove keyring
+  os.unlink(path)
+
+  return True
+
+def test_basic_auth():
+  # make sure we can successfully add/del entities, change their caps
+  # and import/export keyrings.
+
+  expect('ceph auth add client.basicauth', 0)
+  expect('ceph auth caps client.basicauth mon \'allow *\'', 0)
+  # entity exists and caps do not match
+  expect('ceph auth add client.basicauth', errno.EINVAL)
+  # this command attempts to change an existing state and will fail
+  expect('ceph auth add client.basicauth mon \'allow w\'', errno.EINVAL)
+  expect('ceph auth get-or-create client.basicauth', 0)
+  expect('ceph auth get-key client.basicauth', 0)
+  expect('ceph auth get-or-create client.basicauth2', 0)
+  # cleanup
+  expect('ceph auth del client.basicauth', 0)
+  expect('ceph auth del client.basicauth2', 0)
+
+  return True
+
+def gen_module_keyring(module):
+  module_caps = [
+      ('all', '{t} \'allow service {s} rwx\'', 0),
+      ('none', '', errno.EACCES),
+      ('wrong', '{t} \'allow service foobar rwx\'', errno.EACCES),
+      ('right', '{t} \'allow service {s} {p}\'', 0),
+      ('no-execute', '{t} \'allow service {s} x\'', errno.EACCES)
+      ]
+
+  keyring = '{0}.service-{1}'.format(keyring_base,module)
+  for perms in 'r rw x'.split():
+    for (n,p,r) in module_caps:
+      c = p.format(t='mon', s=module, p=perms)
+      expect_to_file(
+          'ceph auth get-or-create client.{cn}-{cp} {caps}'.format(
+          cn=n,cp=perms,caps=c), 0, keyring)
+
+  return keyring
+
+
+def test_all():
+
+
+  perms = {
+      'good': {
+        'broad':[
+          ('rwx', 'allow *'),
+          ('r', 'allow r'),
+          ('rw', 'allow rw'),
+          ('x', 'allow x'),
+          ],
+        'service':[
+          ('rwx', 'allow service {s} rwx'),
+          ('r', 'allow service {s} r'),
+          ('rw', 'allow service {s} rw'),
+          ('x', 'allow service {s} x'),
+          ],
+        'command':[
+          ('rwx', 'allow command "{c}"'),
+          ],
+        'command-with':[
+          ('rwx', 'allow command "{c}" with {kv}')
+          ],
+        'command-with-prefix':[
+          ('rwx', 'allow command "{c}" with {key} prefix {val}')
+          ]
+        },
+      'bad': {
+        'broad':[
+          ('none', ''),
+          ],
+        'service':[
+          ('none1', 'allow service foo rwx'),
+          ('none2', 'allow service foo r'),
+          ('none3', 'allow service foo rw'),
+          ('none4', 'allow service foo x'),
+          ],
+        'command':[
+          ('none', 'allow command foo'),
+          ],
+        'command-with':[
+          ('none', 'allow command "{c}" with foo=bar'),
+          ],
+        'command-with-prefix':[
+          ('none', 'allow command "{c}" with foo prefix bar'),
+          ],
+        }
+      }
+
+  cmds = {
+      '':[
+        {
+          'cmd':('status', '', 'r')
+          },
+        {
+          'pre':'heap start_profiler',
+          'cmd':('heap', 'heapcmd=stats', 'rw'),
+          'post':'heap stop_profiler'
+          }
+        ],
+      'auth':[
+        {
+          'pre':'',
+          'cmd':('auth ls', '', 'r'),
+          'post':''
+          },
+        {
+          'pre':'auth get-or-create client.foo mon \'allow *\'',
+          'cmd':('auth caps', 'entity="client.foo"', 'rw'),
+          'post':'auth del client.foo'
+          }
+        ],
+      'pg':[
+        {
+          'cmd':('pg getmap', '', 'r'),
+          },
+        ],
+      'mds':[
+        {
+          'cmd':('mds getmap', '', 'r'),
+          },
+        ],
+      'mon':[
+        {
+          'cmd':('mon getmap', '', 'r')
+          },
+        {
+          'cmd':('mon remove', 'name=a', 'rw')
+          }
+        ],
+      'osd':[
+        {
+          'cmd':('osd getmap', '', 'r'),
+          },
+        {
+          'cmd':('osd pause', '', 'rw'),
+          'post':'osd unpause'
+          },
+        {
+          'cmd':('osd crush dump', '', 'r')
+          },
+        ],
+      'config-key':[
+          {
+            'pre':'config-key set foo bar',
+            'cmd':('config-key get', 'key=foo', 'r')
+            },
+          {
+            'pre':'config-key set foo bar',
+            'cmd':('config-key del', 'key=foo', 'rw')
+            }
+          ]
+      }
+
+  for (module,cmd_lst) in cmds.items():
+    k = keyring_base + '.' + module
+    for cmd in cmd_lst:
+
+      (cmd_cmd, cmd_args, cmd_perm) = cmd['cmd']
+      cmd_args_key = ''
+      cmd_args_val = ''
+      if len(cmd_args) > 0:
+        (cmd_args_key, cmd_args_val) = cmd_args.split('=')
+
+      print('generating keyring for {m}/{c}'.format(m=module,c=cmd_cmd))
+      # gen keyring
+      for (good_or_bad,kind_map) in perms.items():
+        for (kind,lst) in kind_map.items():
+          for (perm, cap) in lst:
+            cap_formatted = cap.format(
+                s=module,
+                c=cmd_cmd,
+                kv=cmd_args,
+                key=cmd_args_key,
+                val=cmd_args_val)
+
+            if len(cap_formatted) == 0:
+              run_cap = ''
+            else:
+              run_cap = 'mon \'{fc}\''.format(fc=cap_formatted)
+
+            cname = 'client.{gb}-{kind}-{p}'.format(
+                gb=good_or_bad,kind=kind,p=perm)
+            expect_to_file(
+                'ceph auth get-or-create {n} {c}'.format(
+                  n=cname,c=run_cap), 0, k)
+      # keyring generated
+      print('testing {m}/{c}'.format(m=module,c=cmd_cmd))
+
+      # test
+      for good_bad in perms.keys():
+        for (kind,lst) in perms[good_bad].items():
+          for (perm,_) in lst:
+            cname = 'client.{gb}-{k}-{p}'.format(gb=good_bad,k=kind,p=perm)
+
+          if good_bad == 'good':
+            expect_ret = 0
+          else:
+            expect_ret = errno.EACCES
+
+          if ( cmd_perm not in perm ):
+            expect_ret = errno.EACCES
+          if 'with' in kind and len(cmd_args) == 0:
+            expect_ret = errno.EACCES
+          if 'service' in kind and len(module) == 0:
+            expect_ret = errno.EACCES
+
+          if 'pre' in cmd and len(cmd['pre']) > 0:
+            expect('ceph {0}'.format(cmd['pre']), 0)
+          expect('ceph -n {cn} -k {k} {c} {arg_val}'.format(
+            cn=cname,k=k,c=cmd_cmd,arg_val=cmd_args_val), expect_ret)
+          if 'post' in cmd and len(cmd['post']) > 0:
+            expect('ceph {0}'.format(cmd['post']), 0)
+      # finish testing
+      destroy_keyring(k)
+
+
+  return True
+
+
+def test_misc():
+
+  k = keyring_base + '.misc'
+  expect_to_file(
+      'ceph auth get-or-create client.caps mon \'allow command "auth caps"' \
+          ' with entity="client.caps"\'', 0, k)
+  expect('ceph -n client.caps -k {kf} mon_status'.format(kf=k), errno.EACCES)
+  expect('ceph -n client.caps -k {kf} auth caps client.caps mon \'allow *\''.format(kf=k), 0)
+  expect('ceph -n client.caps -k {kf} mon_status'.format(kf=k), 0)
+  destroy_keyring(k)
+
+def main():
+
+  test_basic_auth()
+  test_all()
+  test_misc()
+
+  print('OK')
+
+  return 0
+
+if __name__ == '__main__':
+  main()
diff --git a/qa/workunits/mon/caps.sh b/qa/workunits/mon/caps.sh
new file mode 100755
index 00000000..c5db5650
--- /dev/null
+++ b/qa/workunits/mon/caps.sh
@@ -0,0 +1,90 @@
+#!/usr/bin/env bash
+
+set -x
+
+tmp=/tmp/cephtest-mon-caps-madness
+
+exit_on_error=1
+
+[[ ! -z $TEST_EXIT_ON_ERROR ]] && exit_on_error=$TEST_EXIT_ON_ERROR
+
+if [ `uname` = FreeBSD ]; then
+    ETIMEDOUT=60
+else
+    ETIMEDOUT=110
+fi
+
+expect()
+{
+  cmd=$1
+  expected_ret=$2
+
+  echo $cmd
+  eval $cmd >&/dev/null
+  ret=$?
+
+  if [[ $ret -ne $expected_ret ]]; then
+    echo "Error: Expected return $expected_ret, got $ret"
+    [[ $exit_on_error -eq 1 ]] && exit 1
+    return 1
+  fi
+
+  return 0
+}
+
+expect "ceph auth get-or-create client.bazar > $tmp.bazar.keyring" 0
+expect "ceph -k $tmp.bazar.keyring --user bazar mon_status" 13
+ceph auth del client.bazar
+
+c="'allow command \"auth ls\", allow command mon_status'"
+expect "ceph auth get-or-create client.foo mon $c > $tmp.foo.keyring" 0
+expect "ceph -k $tmp.foo.keyring --user foo mon_status" 0
+expect "ceph -k $tmp.foo.keyring --user foo auth ls" 0
+expect "ceph -k $tmp.foo.keyring --user foo auth export" 13
+expect "ceph -k $tmp.foo.keyring --user foo auth del client.bazar" 13
+expect "ceph -k $tmp.foo.keyring --user foo osd dump" 13
+
+# monitor drops the subscribe message from client if it does not have enough caps
+# for read from mon. in that case, the client will be waiting for mgrmap in vain,
+# if it is instructed to send a command to mgr. "pg dump" is served by mgr. so,
+# we need to set a timeout for testing this scenario.
+#
+# leave plenty of time here because the mons might be thrashing.
+export CEPH_ARGS='--rados-mon-op-timeout=300'
+expect "ceph -k $tmp.foo.keyring --user foo pg dump" $ETIMEDOUT
+export CEPH_ARGS=''
+
+expect "ceph -k $tmp.foo.keyring --user foo quorum_status" 13
+ceph auth del client.foo
+
+c="'allow command service with prefix=list, allow command mon_status'"
+expect "ceph auth get-or-create client.bar mon $c > $tmp.bar.keyring" 0
+expect "ceph -k $tmp.bar.keyring --user bar mon_status" 0
+expect "ceph -k $tmp.bar.keyring --user bar auth ls" 13
+expect "ceph -k $tmp.bar.keyring --user bar auth export" 13
+expect "ceph -k $tmp.bar.keyring --user bar auth del client.foo" 13
+expect "ceph -k $tmp.bar.keyring --user bar osd dump" 13
+
+# again, we'll need to timeout.
+export CEPH_ARGS='--rados-mon-op-timeout=300'
+expect "ceph -k $tmp.bar.keyring --user bar pg dump" $ETIMEDOUT
+export CEPH_ARGS=''
+
+expect "ceph -k $tmp.bar.keyring --user bar quorum_status" 13
+ceph auth del client.bar
+
+rm $tmp.bazar.keyring $tmp.foo.keyring $tmp.bar.keyring
+
+# invalid caps health warning
+cat <<EOF | ceph auth import -i -
+[client.bad]
+  caps mon = this is wrong
+  caps osd = does not parse
+  caps mds = also does not parse
+EOF
+ceph health | grep AUTH_BAD_CAP
+ceph health detail | grep client.bad
+ceph auth rm client.bad
+expect "ceph auth health | grep AUTH_BAD_CAP" 1
+
+echo OK
diff --git a/qa/workunits/mon/config.sh b/qa/workunits/mon/config.sh
new file mode 100755
index 00000000..4f7bdac2
--- /dev/null
+++ b/qa/workunits/mon/config.sh
@@ -0,0 +1,110 @@
+#!/bin/bash -ex
+
+function expect_false()
+{
+	set -x
+	if "$@"; then return 1; else return 0; fi
+}
+
+ceph config dump
+
+# value validation
+ceph config set mon.a debug_xio 22
+ceph config set mon.a debug_xio 22/33
+ceph config get mon.a debug_xio | grep 22
+ceph config set mon.a debug_xio 1/2
+expect_false bin/ceph config set mon.a debug_xio foo
+expect_false bin/ceph config set mon.a debug_xio -10
+ceph config rm mon.a debug_xio
+
+ceph config set global log_graylog_port 123
+expect_false ceph config set global log_graylog_port asdf
+ceph config rm global log_graylog_port
+
+ceph config set mon mon_cluster_log_to_stderr true
+ceph config get mon.a mon_cluster_log_to_stderr | grep true
+ceph config set mon mon_cluster_log_to_stderr 2
+ceph config get mon.a mon_cluster_log_to_stderr | grep true
+ceph config set mon mon_cluster_log_to_stderr 1
+ceph config get mon.a mon_cluster_log_to_stderr | grep true
+ceph config set mon mon_cluster_log_to_stderr false
+ceph config get mon.a mon_cluster_log_to_stderr | grep false
+ceph config set mon mon_cluster_log_to_stderr 0
+ceph config get mon.a mon_cluster_log_to_stderr | grep false
+expect_false ceph config set mon mon_cluster_log_to_stderr fiddle
+expect_false ceph config set mon mon_cluster_log_to_stderr ''
+ceph config rm mon mon_cluster_log_to_stderr
+
+expect_false ceph config set mon.a osd_pool_default_type foo
+ceph config set mon.a osd_pool_default_type replicated
+ceph config rm mon.a osd_pool_default_type
+
+# scoping
+ceph config set global debug_xio 33
+ceph config get mon.a debug_xio | grep 33
+ceph config set mon debug_xio 11
+ceph config get mon.a debug_xio | grep 11
+ceph config set mon.a debug_xio 22
+ceph config get mon.a debug_xio | grep 22
+ceph config rm mon.a debug_xio
+ceph config get mon.a debug_xio | grep 11
+ceph config rm mon debug_xio
+ceph config get mon.a debug_xio | grep 33
+ceph config rm global debug_xio
+
+# help
+ceph config help debug_xio | grep debug_xio
+
+# show
+ceph config set osd.0 debug_xio 33
+while ! ceph config show osd.0 | grep debug_xio | grep 33 | grep mon
+do
+    sleep 1
+done
+ceph config set osd.0 debug_xio 22
+while ! ceph config show osd.0 | grep debug_xio | grep 22 | grep mon
+do
+    sleep 1
+done
+
+ceph tell osd.0 config set debug_xio 99
+while ! ceph config show osd.0 | grep debug_xio | grep 99
+do
+    sleep 1
+done
+ceph config show osd.0 | grep debug_xio | grep 'override mon'
+ceph tell osd.0 config unset debug_xio
+ceph tell osd.0 config unset debug_xio
+
+ceph config rm osd.0 debug_xio
+while ceph config show osd.0 | grep debug_xio | grep mon
+do
+    sleep 1
+done
+ceph config show osd.0 | grep -c debug_xio | grep 0
+
+ceph config set osd.0 osd_scrub_cost 123
+while ! ceph config show osd.0 | grep osd_scrub_cost | grep mon
+do
+    sleep 1
+done
+ceph config rm osd.0 osd_scrub_cost
+
+# show-with-defaults
+ceph config show-with-defaults osd.0 | grep debug_xio
+
+# assimilate
+t1=`mktemp`
+t2=`mktemp`
+cat <<EOF > $t1
+[osd.0]
+keyring = foo
+debug_xio = 66
+EOF
+ceph config assimilate-conf -i $t1 | tee $t2
+
+grep keyring $t2
+expect_false grep debug_xio $t2
+rm -f $t1 $t2
+
+echo OK
diff --git a/qa/workunits/mon/crush_ops.sh b/qa/workunits/mon/crush_ops.sh
new file mode 100755
index 00000000..4ad8f354
--- /dev/null
+++ b/qa/workunits/mon/crush_ops.sh
@@ -0,0 +1,239 @@
+#!/usr/bin/env bash
+
+set -ex
+
+function expect_false()
+{
+	set -x
+	if "$@"; then return 1; else return 0; fi
+}
+
+ceph osd crush dump
+
+# rules
+ceph osd crush rule dump
+ceph osd crush rule ls
+ceph osd crush rule list
+
+ceph osd crush rule create-simple foo default host
+ceph osd crush rule create-simple foo default host
+ceph osd crush rule create-simple bar default host
+
+# make sure we're at luminous+ before using crush device classes
+ceph osd require-osd-release nautilus
+ceph osd crush rm-device-class all
+ceph osd crush set-device-class ssd osd.0
+ceph osd crush set-device-class hdd osd.1
+ceph osd crush rule create-replicated foo-ssd default host ssd
+ceph osd crush rule create-replicated foo-hdd default host hdd
+ceph osd crush rule ls-by-class ssd | grep 'foo-ssd'
+ceph osd crush rule ls-by-class ssd | expect_false grep 'foo-hdd'
+ceph osd crush rule ls-by-class hdd | grep 'foo-hdd'
+ceph osd crush rule ls-by-class hdd | expect_false grep 'foo-ssd'
+
+ceph osd erasure-code-profile set ec-foo-ssd crush-device-class=ssd m=2 k=2
+ceph osd pool create ec-foo 2 erasure ec-foo-ssd
+ceph osd pool rm ec-foo ec-foo --yes-i-really-really-mean-it
+
+ceph osd crush rule ls | grep foo
+
+ceph osd crush rule rename foo foo-asdf
+ceph osd crush rule rename foo foo-asdf # idempotent
+ceph osd crush rule rename bar bar-asdf
+ceph osd crush rule ls | grep 'foo-asdf'
+ceph osd crush rule ls | grep 'bar-asdf'
+ceph osd crush rule rm foo 2>&1 | grep 'does not exist'
+ceph osd crush rule rm bar 2>&1 | grep 'does not exist'
+ceph osd crush rule rename foo-asdf foo
+ceph osd crush rule rename foo-asdf foo # idempotent
+ceph osd crush rule rename bar-asdf bar
+ceph osd crush rule ls | expect_false grep 'foo-asdf'
+ceph osd crush rule ls | expect_false grep 'bar-asdf'
+ceph osd crush rule rm foo
+ceph osd crush rule rm foo  # idempotent
+ceph osd crush rule rm bar
+
+# can't delete in-use rules, tho:
+ceph osd pool create pinning_pool 1
+expect_false ceph osd crush rule rm replicated_rule
+ceph osd pool rm pinning_pool pinning_pool --yes-i-really-really-mean-it
+
+# build a simple map
+expect_false ceph osd crush add-bucket foo osd
+ceph osd crush add-bucket foo root
+o1=`ceph osd create`
+o2=`ceph osd create`
+ceph osd crush add $o1 1 host=host1 root=foo
+ceph osd crush add $o1 1 host=host1 root=foo  # idemptoent
+ceph osd crush add $o2 1 host=host2 root=foo
+ceph osd crush add $o2 1 host=host2 root=foo  # idempotent
+ceph osd crush add-bucket bar root
+ceph osd crush add-bucket bar root  # idempotent
+ceph osd crush link host1 root=bar
+ceph osd crush link host1 root=bar  # idempotent
+ceph osd crush link host2 root=bar
+ceph osd crush link host2 root=bar  # idempotent
+
+ceph osd tree | grep -c osd.$o1 | grep -q 2
+ceph osd tree | grep -c host1 | grep -q 2
+ceph osd tree | grep -c osd.$o2 | grep -q 2
+ceph osd tree | grep -c host2 | grep -q 2
+expect_false ceph osd crush rm host1 foo   # not empty
+ceph osd crush unlink host1 foo
+ceph osd crush unlink host1 foo
+ceph osd tree | grep -c host1 | grep -q 1
+
+expect_false ceph osd crush rm foo  # not empty
+expect_false ceph osd crush rm bar  # not empty
+ceph osd crush unlink host1 bar
+ceph osd tree | grep -c host1 | grep -q 1   # now an orphan
+ceph osd crush rm osd.$o1 host1
+ceph osd crush rm host1
+ceph osd tree | grep -c host1 | grep -q 0
+expect_false ceph osd tree-from host1
+ceph osd tree-from host2
+expect_false ceph osd tree-from osd.$o2
+
+expect_false ceph osd crush rm bar   # not empty
+ceph osd crush unlink host2
+
+ceph osd crush add-bucket host-for-test host root=root-for-test rack=rack-for-test
+ceph osd tree | grep host-for-test
+ceph osd tree | grep rack-for-test
+ceph osd tree | grep root-for-test
+ceph osd crush rm host-for-test
+ceph osd crush rm rack-for-test
+ceph osd crush rm root-for-test
+
+# reference foo and bar with a rule
+ceph osd crush rule create-simple foo-rule foo host firstn
+expect_false ceph osd crush rm foo
+ceph osd crush rule rm foo-rule
+
+ceph osd crush rm bar
+ceph osd crush rm foo
+ceph osd crush rm osd.$o2 host2
+ceph osd crush rm host2
+
+ceph osd crush add-bucket foo host
+ceph osd crush move foo root=default rack=localrack
+
+ceph osd crush create-or-move osd.$o1 1.0 root=default
+ceph osd crush move osd.$o1 host=foo
+ceph osd find osd.$o1 | grep host | grep foo
+
+ceph osd crush rm osd.$o1
+ceph osd crush rm osd.$o2
+
+ceph osd crush rm foo
+
+# test reweight
+o3=`ceph osd create`
+ceph osd crush add $o3 123 root=default
+ceph osd tree | grep osd.$o3 | grep 123
+ceph osd crush reweight osd.$o3 113
+expect_false ceph osd crush reweight osd.$o3 123456
+ceph osd tree | grep osd.$o3 | grep 113
+ceph osd crush rm osd.$o3
+ceph osd rm osd.$o3
+
+# test reweight-subtree
+o4=`ceph osd create`
+o5=`ceph osd create`
+ceph osd crush add $o4 123 root=default host=foobaz
+ceph osd crush add $o5 123 root=default host=foobaz
+ceph osd tree | grep osd.$o4 | grep 123
+ceph osd tree | grep osd.$o5 | grep 123
+ceph osd crush reweight-subtree foobaz 155
+expect_false ceph osd crush reweight-subtree foobaz 123456
+ceph osd tree | grep osd.$o4 | grep 155
+ceph osd tree | grep osd.$o5 | grep 155
+ceph osd crush rm osd.$o4
+ceph osd crush rm osd.$o5
+ceph osd rm osd.$o4
+ceph osd rm osd.$o5
+
+# weight sets
+# make sure we require luminous before testing weight-sets
+ceph osd set-require-min-compat-client luminous
+ceph osd crush weight-set dump
+ceph osd crush weight-set ls
+expect_false ceph osd crush weight-set reweight fooset osd.0 .9
+ceph osd pool create fooset 8
+ceph osd pool create barset 8
+ceph osd pool set barset size 3
+expect_false ceph osd crush weight-set reweight fooset osd.0 .9
+ceph osd crush weight-set create fooset flat
+ceph osd crush weight-set create barset positional
+ceph osd crush weight-set ls | grep fooset
+ceph osd crush weight-set ls | grep barset
+ceph osd crush weight-set dump
+ceph osd crush weight-set reweight fooset osd.0 .9
+expect_false ceph osd crush weight-set reweight fooset osd.0 .9 .9
+expect_false ceph osd crush weight-set reweight barset osd.0 .9
+ceph osd crush weight-set reweight barset osd.0 .9 .9 .9
+ceph osd crush weight-set ls | grep -c fooset | grep -q 1
+ceph osd crush weight-set rm fooset
+ceph osd crush weight-set ls | grep -c fooset | grep -q 0
+ceph osd crush weight-set ls | grep barset
+ceph osd crush weight-set rm barset
+ceph osd crush weight-set ls | grep -c barset | grep -q 0
+ceph osd crush weight-set create-compat
+ceph osd crush weight-set ls | grep '(compat)'
+ceph osd crush weight-set rm-compat
+
+# weight set vs device classes
+ceph osd pool create cool 2
+ceph osd pool create cold 2
+ceph osd pool set cold size 2
+ceph osd crush weight-set create-compat
+ceph osd crush weight-set create cool flat
+ceph osd crush weight-set create cold positional
+ceph osd crush rm-device-class osd.0
+ceph osd crush weight-set reweight-compat osd.0 10.5
+ceph osd crush weight-set reweight cool osd.0 11.5
+ceph osd crush weight-set reweight cold osd.0 12.5 12.4
+ceph osd crush set-device-class fish osd.0
+ceph osd crush tree --show-shadow | grep osd\\.0 | grep fish | grep 10\\.
+ceph osd crush tree --show-shadow | grep osd\\.0 | grep fish | grep 11\\.
+ceph osd crush tree --show-shadow | grep osd\\.0 | grep fish | grep 12\\.
+ceph osd crush rm-device-class osd.0
+ceph osd crush set-device-class globster osd.0
+ceph osd crush tree --show-shadow | grep osd\\.0 | grep globster | grep 10\\.
+ceph osd crush tree --show-shadow | grep osd\\.0 | grep globster | grep 11\\.
+ceph osd crush tree --show-shadow | grep osd\\.0 | grep globster | grep 12\\.
+ceph osd crush weight-set reweight-compat osd.0 7.5
+ceph osd crush weight-set reweight cool osd.0 8.5
+ceph osd crush weight-set reweight cold osd.0 6.5 6.6
+ceph osd crush tree --show-shadow | grep osd\\.0 | grep globster | grep 7\\.
+ceph osd crush tree --show-shadow | grep osd\\.0 | grep globster | grep 8\\.
+ceph osd crush tree --show-shadow | grep osd\\.0 | grep globster | grep 6\\.
+ceph osd crush rm-device-class osd.0
+ceph osd pool rm cool cool --yes-i-really-really-mean-it
+ceph osd pool rm cold cold --yes-i-really-really-mean-it
+ceph osd crush weight-set rm-compat
+
+# weight set vs device classes vs move
+ceph osd crush weight-set create-compat
+ceph osd crush add-bucket fooo host
+ceph osd crush move fooo root=default
+ceph osd crush add-bucket barr rack
+ceph osd crush move barr root=default
+ceph osd crush move fooo rack=barr
+ceph osd crush rm fooo
+ceph osd crush rm barr
+ceph osd crush weight-set rm-compat
+
+# this sequence would crash at one point
+ceph osd crush weight-set create-compat
+ceph osd crush add-bucket r1 rack root=default
+for f in `seq 1 32`; do
+    ceph osd crush add-bucket h$f host rack=r1
+done
+for f in `seq 1 32`; do
+    ceph osd crush rm h$f
+done
+ceph osd crush rm r1
+ceph osd crush weight-set rm-compat
+
+echo OK
diff --git a/qa/workunits/mon/osd.sh b/qa/workunits/mon/osd.sh
new file mode 100755
index 00000000..535d6c13
--- /dev/null
+++ b/qa/workunits/mon/osd.sh
@@ -0,0 +1,24 @@
+#!/bin/sh -x
+
+set -e
+
+ua=`uuidgen`
+ub=`uuidgen`
+
+# should get same id with same uuid
+na=`ceph osd create $ua`
+test $na -eq `ceph osd create $ua`
+
+nb=`ceph osd create $ub`
+test $nb -eq `ceph osd create $ub`
+test $nb -ne $na
+
+ceph osd rm $na
+ceph osd rm $na
+ceph osd rm $nb
+ceph osd rm 1000
+
+na2=`ceph osd create $ua`
+
+echo OK
+
diff --git a/qa/workunits/mon/pg_autoscaler.sh b/qa/workunits/mon/pg_autoscaler.sh
new file mode 100755
index 00000000..706f87d0
--- /dev/null
+++ b/qa/workunits/mon/pg_autoscaler.sh
@@ -0,0 +1,79 @@
+#!/bin/bash -ex
+
+NUM_OSDS=$(ceph osd ls | wc -l)
+if [ $NUM_OSDS -lt 6 ]; then
+    echo "test requires at least 6 OSDs"
+    exit 1
+fi
+
+NUM_POOLS=$(ceph osd pool ls | wc -l)
+if [ $NUM_POOLS -gt 0 ]; then
+    echo "test requires no preexisting pools"
+    exit 1
+fi
+
+function wait_for() {
+    local sec=$1
+    local cmd=$2
+
+    while true ; do
+	if bash -c "$cmd" ; then
+	    break
+	fi
+	sec=$(( $sec - 1 ))
+	if [ $sec -eq 0 ]; then
+	    echo failed
+	    return 1
+	fi
+	sleep 1
+    done
+    return 0
+}
+
+# enable
+ceph config set mgr mgr/pg_autoscaler/sleep_interval 5
+ceph mgr module enable pg_autoscaler
+
+# pg_num_min
+ceph osd pool create a 16 --pg-num-min 4
+ceph osd pool create b 16 --pg-num-min 2
+ceph osd pool set a pg_autoscale_mode on
+ceph osd pool set b pg_autoscale_mode on
+
+wait_for 120 "ceph osd pool get a pg_num | grep 4"
+wait_for 120 "ceph osd pool get b pg_num | grep 2"
+
+# target ratio
+ceph osd pool set a target_size_ratio 5
+ceph osd pool set b target_size_ratio 1
+sleep 10
+APGS=$(ceph osd dump -f json-pretty | jq '.pools[0].pg_num_target')
+BPGS=$(ceph osd dump -f json-pretty | jq '.pools[1].pg_num_target')
+test $APGS -gt 100
+test $BPGS -gt 10
+
+# small ratio change does not change pg_num
+ceph osd pool set a target_size_ratio 7
+ceph osd pool set b target_size_ratio 2
+sleep 10
+APGS2=$(ceph osd dump -f json-pretty | jq '.pools[0].pg_num_target')
+BPGS2=$(ceph osd dump -f json-pretty | jq '.pools[1].pg_num_target')
+test $APGS -eq $APGS2
+test $BPGS -eq $BPGS2
+
+# target_size
+ceph osd pool set a target_size_bytes 1000000000000000
+ceph osd pool set b target_size_bytes 1000000000000000
+ceph osd pool set a target_size_ratio 0
+ceph osd pool set b target_size_ratio 0
+wait_for 60 "ceph health detail | grep POOL_TARGET_SIZE_BYTES_OVERCOMMITTED"
+
+ceph osd pool set a target_size_bytes 1000
+ceph osd pool set b target_size_bytes 1000
+ceph osd pool set a target_size_ratio 1
+wait_for 60 "ceph health detail | grep POOL_HAS_TARGET_SIZE_BYTES_AND_RATIO"
+
+ceph osd pool rm a a --yes-i-really-really-mean-it
+ceph osd pool rm b b --yes-i-really-really-mean-it
+
+echo OK
diff --git a/qa/workunits/mon/ping.py b/qa/workunits/mon/ping.py
new file mode 100755
index 00000000..f39da885
--- /dev/null
+++ b/qa/workunits/mon/ping.py
@@ -0,0 +1,108 @@
+#!/usr/bin/python
+
+import json
+import shlex
+import subprocess
+
+import six
+
+
+class UnexpectedReturn(Exception):
+    def __init__(self, cmd, ret, expected, msg):
+        if isinstance(cmd, list):
+            self.cmd = ' '.join(cmd)
+        else:
+            assert isinstance(cmd, six.string_types) or isinstance(cmd, six.text_type), \
+                'cmd needs to be either a list or a str'
+            self.cmd = cmd
+        self.cmd = str(self.cmd)
+        self.ret = int(ret)
+        self.expected = int(expected)
+        self.msg = str(msg)
+
+    def __str__(self):
+        return repr('{c}: expected return {e}, got {r} ({o})'.format(
+            c=self.cmd, e=self.expected, r=self.ret, o=self.msg))
+
+
+def call(cmd):
+    if isinstance(cmd, list):
+        args = cmd
+    elif isinstance(cmd, six.string_types) or isinstance(cmd, six.text_type):
+        args = shlex.split(cmd)
+    else:
+        assert False, 'cmd is not a string/unicode nor a list!'
+
+    print('call: {0}'.format(args))
+    proc = subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    procout, procerr = proc.communicate(None)
+
+    return proc.returncode, procout, procerr
+
+
+def expect(cmd, expected_ret):
+    try:
+        (r, out, err) = call(cmd)
+    except ValueError as e:
+        assert False, \
+            'unable to run {c}: {err}'.format(c=repr(cmd), err=str(e))
+
+    if r != expected_ret:
+        raise UnexpectedReturn(repr(cmd), r, expected_ret, err)
+
+    return out.decode() if isinstance(out, bytes) else out
+
+
+def get_quorum_status(timeout=300):
+    cmd = 'ceph quorum_status'
+    if timeout > 0:
+        cmd += ' --connect-timeout {0}'.format(timeout)
+
+    out = expect(cmd, 0)
+    j = json.loads(out)
+    return j
+
+
+def main():
+    quorum_status = get_quorum_status()
+    mon_names = [mon['name'] for mon in quorum_status['monmap']['mons']]
+
+    print('ping all monitors')
+    for m in mon_names:
+        print('ping mon.{0}'.format(m))
+        out = expect('ceph ping mon.{0}'.format(m), 0)
+        reply = json.loads(out)
+
+        assert reply['mon_status']['name'] == m, \
+            'reply obtained from mon.{0}, expected mon.{1}'.format(
+                reply['mon_status']['name'], m)
+
+    print('test out-of-quorum reply')
+    for m in mon_names:
+        print('testing mon.{0}'.format(m))
+        expect('ceph daemon mon.{0} quorum exit'.format(m), 0)
+
+        quorum_status = get_quorum_status()
+        assert m not in quorum_status['quorum_names'], \
+            'mon.{0} was not supposed to be in quorum ({1})'.format(
+                m, quorum_status['quorum_names'])
+
+        out = expect('ceph ping mon.{0}'.format(m), 0)
+        reply = json.loads(out)
+        mon_status = reply['mon_status']
+
+        assert mon_status['name'] == m, \
+            'reply obtained from mon.{0}, expected mon.{1}'.format(
+                mon_status['name'], m)
+
+        assert mon_status['state'] == 'electing', \
+            'mon.{0} is in state {1}, expected electing'.format(
+                m, mon_status['state'])
+
+        expect('ceph daemon mon.{0} quorum enter'.format(m), 0)
+
+    print('OK')
+
+
+if __name__ == '__main__':
+    main()
diff --git a/qa/workunits/mon/pool_ops.sh b/qa/workunits/mon/pool_ops.sh
new file mode 100755
index 00000000..b0207769
--- /dev/null
+++ b/qa/workunits/mon/pool_ops.sh
@@ -0,0 +1,97 @@
+#!/usr/bin/env bash
+
+set -ex
+
+function expect_false()
+{
+	set -x
+	if "$@"; then return 1; else return 0; fi
+}
+
+function get_config_value_or_die()
+{
+  local pool_name config_opt raw val
+
+  pool_name=$1
+  config_opt=$2
+
+  raw="`$SUDO ceph osd pool get $pool_name $config_opt 2>/dev/null`"
+  if [[ $? -ne 0 ]]; then
+    echo "error obtaining config opt '$config_opt' from '$pool_name': $raw"
+    exit 1
+  fi
+
+  raw=`echo $raw | sed -e 's/[{} "]//g'`
+  val=`echo $raw | cut -f2 -d:`
+
+  echo "$val"
+  return 0
+}
+
+function expect_config_value()
+{
+  local pool_name config_opt expected_val val
+  pool_name=$1
+  config_opt=$2
+  expected_val=$3
+
+  val=$(get_config_value_or_die $pool_name $config_opt)
+
+  if [[ "$val" != "$expected_val" ]]; then
+    echo "expected '$expected_val', got '$val'"
+    exit 1
+  fi
+}
+
+# note: we need to pass the other args or ceph_argparse.py will take
+# 'invalid' that is not replicated|erasure and assume it is the next
+# argument, which is a string.
+expect_false ceph osd pool create foo 123 123 invalid foo-profile foo-ruleset
+
+ceph osd pool create foo 123 123 replicated
+ceph osd pool create fooo 123 123 erasure default
+ceph osd pool create foooo 123
+
+ceph osd pool create foo 123 # idempotent
+
+ceph osd pool set foo size 1
+expect_config_value "foo" "min_size" 1
+ceph osd pool set foo size 4
+expect_config_value "foo" "min_size" 2
+ceph osd pool set foo size 10
+expect_config_value "foo" "min_size" 5
+expect_false ceph osd pool set foo size 0
+expect_false ceph osd pool set foo size 20
+
+ceph osd pool set foo size 3
+ceph osd getcrushmap -o crush
+crushtool -d crush -o crush.txt
+sed -i 's/max_size 10/max_size 3/' crush.txt
+crushtool -c crush.txt -o crush.new
+ceph osd setcrushmap -i crush.new
+expect_false ceph osd pool set foo size 4
+ceph osd setcrushmap -i crush
+rm -f crush crush.txt crush.new
+
+# should fail due to safety interlock
+expect_false ceph osd pool delete foo
+expect_false ceph osd pool delete foo foo
+expect_false ceph osd pool delete foo foo --force
+expect_false ceph osd pool delete foo fooo --yes-i-really-mean-it
+expect_false ceph osd pool delete foo --yes-i-really-mean-it foo
+
+ceph osd pool delete foooo foooo --yes-i-really-really-mean-it
+ceph osd pool delete fooo fooo --yes-i-really-really-mean-it
+ceph osd pool delete foo foo --yes-i-really-really-mean-it
+
+# idempotent
+ceph osd pool delete foo foo --yes-i-really-really-mean-it
+ceph osd pool delete fooo fooo --yes-i-really-really-mean-it
+ceph osd pool delete fooo fooo --yes-i-really-really-mean-it
+
+# non-existent pool
+ceph osd pool delete fuggg fuggg --yes-i-really-really-mean-it
+
+echo OK
+
+
diff --git a/qa/workunits/mon/rbd_snaps_ops.sh b/qa/workunits/mon/rbd_snaps_ops.sh
new file mode 100755
index 00000000..eb88565e
--- /dev/null
+++ b/qa/workunits/mon/rbd_snaps_ops.sh
@@ -0,0 +1,61 @@
+#!/usr/bin/env bash
+
+# attempt to trigger #6047
+
+
+cmd_no=0
+expect()
+{
+  cmd_no=$(($cmd_no+1))
+  cmd="$1"
+  expected=$2
+  echo "[$cmd_no] $cmd"
+  eval  $cmd
+  ret=$?
+  if [[ $ret -ne $expected ]]; then
+    echo "[$cmd_no] unexpected return '$ret', expected '$expected'"
+    exit 1
+  fi
+}
+
+ceph osd pool delete test test --yes-i-really-really-mean-it || true
+expect 'ceph osd pool create test 8 8' 0
+expect 'ceph osd pool application enable test rbd'
+expect 'ceph osd pool mksnap test snapshot' 0
+expect 'ceph osd pool rmsnap test snapshot' 0
+
+expect 'rbd --pool=test --rbd_validate_pool=false create --size=102400 image' 0
+expect 'rbd --pool=test snap create image@snapshot' 22
+
+expect 'ceph osd pool delete test test --yes-i-really-really-mean-it' 0
+expect 'ceph osd pool create test 8 8' 0
+expect 'rbd --pool=test pool init' 0
+expect 'rbd --pool=test create --size=102400 image' 0
+expect 'rbd --pool=test snap create image@snapshot' 0
+expect 'rbd --pool=test snap ls image' 0
+expect 'rbd --pool=test snap rm image@snapshot' 0
+
+expect 'ceph osd pool mksnap test snapshot' 22
+
+expect 'ceph osd pool delete test test --yes-i-really-really-mean-it' 0
+
+# reproduce 7210 and expect it to be fixed
+# basically create such a scenario where we end up deleting what used to
+# be an unmanaged snapshot from a not-unmanaged pool
+
+ceph osd pool delete test-foo test-foo --yes-i-really-really-mean-it || true
+expect 'ceph osd pool create test-foo 8' 0
+expect 'ceph osd pool application enable test-foo rbd'
+expect 'rbd --pool test-foo create --size 1024 image' 0
+expect 'rbd --pool test-foo snap create image@snapshot' 0
+
+ceph osd pool delete test-bar test-bar --yes-i-really-really-mean-it || true
+expect 'ceph osd pool create test-bar 8' 0
+expect 'ceph osd pool application enable test-bar rbd'
+expect 'rados cppool test-foo test-bar --yes-i-really-mean-it' 0
+expect 'rbd --pool test-bar snap rm image@snapshot' 95
+expect 'ceph osd pool delete test-foo test-foo --yes-i-really-really-mean-it' 0
+expect 'ceph osd pool delete test-bar test-bar --yes-i-really-really-mean-it' 0
+
+
+echo OK
diff --git a/qa/workunits/mon/test_config_key_caps.sh b/qa/workunits/mon/test_config_key_caps.sh
new file mode 100755
index 00000000..77b4b53b
--- /dev/null
+++ b/qa/workunits/mon/test_config_key_caps.sh
@@ -0,0 +1,201 @@
+#!/usr/bin/env bash
+
+set -x
+set -e
+
+tmp=$(mktemp -d -p /tmp test_mon_config_key_caps.XXXXX)
+entities=()
+
+function cleanup()
+{
+	set +e
+	set +x
+	if [[ -e $tmp/keyring ]] && [[ -e $tmp/keyring.orig ]]; then
+		grep '\[.*\..*\]' $tmp/keyring.orig > $tmp/entities.orig
+		for e in $(grep '\[.*\..*\]' $tmp/keyring | \
+			diff $tmp/entities.orig - | \
+			sed -n 's/^.*\[\(.*\..*\)\]/\1/p');
+		do
+			ceph auth rm $e 2>&1 >& /dev/null
+		done
+	fi
+	#rm -fr $tmp
+}
+
+trap cleanup 0 # cleanup on exit
+
+function expect_false()
+{
+	set -x
+	if "$@"; then return 1; else return 0; fi
+}
+
+# for cleanup purposes
+ceph auth export -o $tmp/keyring.orig
+
+k=$tmp/keyring
+
+# setup a few keys
+ceph config-key ls
+ceph config-key set daemon-private/osd.123/test-foo
+ceph config-key set mgr/test-foo
+ceph config-key set device/test-foo
+ceph config-key set test/foo
+
+allow_aa=client.allow_aa
+allow_bb=client.allow_bb
+allow_cc=client.allow_cc
+
+mgr_a=mgr.a
+mgr_b=mgr.b
+osd_a=osd.100
+osd_b=osd.200
+
+prefix_aa=client.prefix_aa
+prefix_bb=client.prefix_bb
+prefix_cc=client.prefix_cc
+match_aa=client.match_aa
+match_bb=client.match_bb
+
+fail_aa=client.fail_aa
+fail_bb=client.fail_bb
+fail_cc=client.fail_cc
+fail_dd=client.fail_dd
+fail_ee=client.fail_ee
+fail_ff=client.fail_ff
+fail_gg=client.fail_gg
+fail_writes=client.fail_writes
+
+ceph auth get-or-create $allow_aa mon 'allow *'
+ceph auth get-or-create $allow_bb mon 'allow service config-key rwx'
+ceph auth get-or-create $allow_cc mon 'allow command "config-key get"'
+
+ceph auth get-or-create $mgr_a mon 'allow profile mgr'
+ceph auth get-or-create $mgr_b mon 'allow profile mgr'
+ceph auth get-or-create $osd_a mon 'allow profile osd'
+ceph auth get-or-create $osd_b mon 'allow profile osd'
+
+ceph auth get-or-create $prefix_aa mon \
+	"allow command \"config-key get\" with key prefix client/$prefix_aa"
+
+cap="allow command \"config-key set\" with key prefix client/"
+cap="$cap,allow command \"config-key get\" with key prefix client/$prefix_bb"
+ceph auth get-or-create $prefix_bb mon "$cap"
+
+cap="allow command \"config-key get\" with key prefix client/"
+cap="$cap, allow command \"config-key set\" with key prefix client/"
+cap="$cap, allow command \"config-key ls\""
+ceph auth get-or-create $prefix_cc mon "$cap"
+
+cap="allow command \"config-key get\" with key=client/$match_aa/foo"
+ceph auth get-or-create $match_aa mon "$cap"
+cap="allow command \"config-key get\" with key=client/$match_bb/foo"
+cap="$cap,allow command \"config-key set\" with key=client/$match_bb/foo"
+ceph auth get-or-create $match_bb mon "$cap"
+
+ceph auth get-or-create $fail_aa mon 'allow rx'
+ceph auth get-or-create $fail_bb mon 'allow r,allow w'
+ceph auth get-or-create $fail_cc mon 'allow rw'
+ceph auth get-or-create $fail_dd mon 'allow rwx'
+ceph auth get-or-create $fail_ee mon 'allow profile bootstrap-rgw'
+ceph auth get-or-create $fail_ff mon 'allow profile bootstrap-rbd'
+# write commands will require rw; wx is not enough
+ceph auth get-or-create $fail_gg mon 'allow service config-key wx'
+# read commands will only require 'r'; 'rx' should be enough.
+ceph auth get-or-create $fail_writes mon 'allow service config-key rx'
+
+# grab keyring
+ceph auth export -o $k
+
+# keys will all the caps can do whatever
+for c in $allow_aa $allow_bb $allow_cc $mgr_a $mgr_b; do
+	ceph -k $k --name $c config-key get daemon-private/osd.123/test-foo
+	ceph -k $k --name $c config-key get mgr/test-foo
+	ceph -k $k --name $c config-key get device/test-foo
+	ceph -k $k --name $c config-key get test/foo
+done
+
+for c in $osd_a $osd_b; do
+	ceph -k $k --name $c config-key put daemon-private/$c/test-foo
+	ceph -k $k --name $c config-key get daemon-private/$c/test-foo
+	expect_false ceph -k $k --name $c config-key ls
+	expect_false ceph -k $k --name $c config-key get mgr/test-foo
+	expect_false ceph -k $k --name $c config-key get device/test-foo
+	expect_false ceph -k $k --name $c config-key get test/foo
+done
+
+expect_false ceph -k $k --name $osd_a get daemon-private/$osd_b/test-foo
+expect_false ceph -k $k --name $osd_b get daemon-private/$osd_a/test-foo
+
+expect_false ceph -k $k --name $prefix_aa \
+	config-key ls
+expect_false ceph -k $k --name $prefix_aa \
+	config-key get daemon-private/osd.123/test-foo
+expect_false ceph -k $k --name $prefix_aa \
+	config-key set test/bar
+expect_false ceph -k $k --name $prefix_aa \
+	config-key set client/$prefix_aa/foo
+
+# write something so we can read, use a custom entity
+ceph -k $k --name $allow_bb config-key set client/$prefix_aa/foo
+ceph -k $k --name $prefix_aa config-key get client/$prefix_aa/foo
+# check one writes to the other's prefix, the other is able to read
+ceph -k $k --name $prefix_bb config-key set client/$prefix_aa/bar
+ceph -k $k --name $prefix_aa config-key get client/$prefix_aa/bar
+
+ceph -k $k --name $prefix_bb config-key set client/$prefix_bb/foo
+ceph -k $k --name $prefix_bb config-key get client/$prefix_bb/foo
+
+expect_false ceph -k $k --name $prefix_bb config-key get client/$prefix_aa/bar
+expect_false ceph -k $k --name $prefix_bb config-key ls
+expect_false ceph -k $k --name $prefix_bb \
+	config-key get daemon-private/osd.123/test-foo
+expect_false ceph -k $k --name $prefix_bb config-key get mgr/test-foo
+expect_false ceph -k $k --name $prefix_bb config-key get device/test-foo
+expect_false ceph -k $k --name $prefix_bb config-key get test/bar
+expect_false ceph -k $k --name $prefix_bb config-key set test/bar
+
+ceph -k $k --name $prefix_cc config-key set client/$match_aa/foo
+ceph -k $k --name $prefix_cc config-key set client/$match_bb/foo
+ceph -k $k --name $prefix_cc config-key get client/$match_aa/foo
+ceph -k $k --name $prefix_cc config-key get client/$match_bb/foo
+expect_false ceph -k $k --name $prefix_cc config-key set other/prefix
+expect_false ceph -k $k --name $prefix_cc config-key get mgr/test-foo
+ceph -k $k --name $prefix_cc config-key ls >& /dev/null
+
+ceph -k $k --name $match_aa config-key get client/$match_aa/foo
+expect_false ceph -k $k --name $match_aa config-key get client/$match_bb/foo
+expect_false ceph -k $k --name $match_aa config-key set client/$match_aa/foo
+ceph -k $k --name $match_bb config-key get client/$match_bb/foo
+ceph -k $k --name $match_bb config-key set client/$match_bb/foo
+expect_false ceph -k $k --name $match_bb config-key get client/$match_aa/foo
+expect_false ceph -k $k --name $match_bb config-key set client/$match_aa/foo
+
+keys=(daemon-private/osd.123/test-foo
+	  mgr/test-foo
+	  device/test-foo
+	  test/foo
+	  client/$prefix_aa/foo
+	  client/$prefix_bb/foo
+	  client/$match_aa/foo
+	  client/$match_bb/foo
+)
+# expect these all to fail accessing config-key
+for c in $fail_aa $fail_bb $fail_cc \
+		 $fail_dd $fail_ee $fail_ff \
+		 $fail_gg; do
+	for m in get set; do
+		for key in ${keys[*]} client/$prefix_aa/foo client/$prefix_bb/foo; do
+			expect_false ceph -k $k --name $c config-key $m $key
+		done
+	done
+done
+
+# fail writes but succeed on reads
+expect_false ceph -k $k --name $fail_writes config-key set client/$match_aa/foo
+expect_false ceph -k $k --name $fail_writes config-key set test/foo
+ceph -k $k --name $fail_writes config-key ls
+ceph -k $k --name $fail_writes config-key get client/$match_aa/foo 
+ceph -k $k --name $fail_writes config-key get daemon-private/osd.123/test-foo
+
+echo "OK"
diff --git a/qa/workunits/mon/test_mon_config_key.py b/qa/workunits/mon/test_mon_config_key.py
new file mode 100755
index 00000000..c0cb8299
--- /dev/null
+++ b/qa/workunits/mon/test_mon_config_key.py
@@ -0,0 +1,481 @@
+#!/usr/bin/python
+#
+# test_mon_config_key - Test 'ceph config-key' interface
+#
+# Copyright (C) 2013 Inktank
+#
+# This is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License version 2.1, as published by the Free Software
+# Foundation.  See file COPYING.
+#
+import argparse
+import base64
+import errno
+import json
+import logging
+import os
+import random
+import string
+import subprocess
+import sys
+import time
+
+#
+# Accepted Environment variables:
+#   CEPH_TEST_VERBOSE     - be more verbose; '1' enables; '0' disables
+#   CEPH_TEST_DURATION    - test duration in seconds
+#   CEPH_TEST_SEED        - seed to be used during the test
+#
+# Accepted arguments and options (see --help):
+#   -v, --verbose         - be more verbose
+#   -d, --duration SECS   - test duration in seconds
+#   -s, --seed SEED       - seed to be used during the test
+#
+
+
+LOG = logging.getLogger(os.path.basename(sys.argv[0].replace('.py', '')))
+
+SIZES = [
+    (0, 0),
+    (10, 0),
+    (25, 0),
+    (50, 0),
+    (100, 0),
+    (1000, 0),
+    (64 * 1024, 0),
+    (64 * 1024 + 1, -errno.EFBIG),
+    (128 * 1024, -errno.EFBIG)
+]
+
+# tests will be randomly selected from the keys here, and the test
+# suboperation will be randomly selected from the list in the values
+# here.  i.e. 'exists/existing' would test that a key the test put into
+# the store earlier actually does still exist in the config store,
+# and that's a separate test case from 'exists/enoent', which tests
+# nonexistence of a key known to not be present.
+
+OPS = {
+    'put': ['existing', 'new'],
+    'del': ['existing', 'enoent'],
+    'exists': ['existing', 'enoent'],
+    'get': ['existing', 'enoent'],
+    'list': ['existing', 'enoent'],
+    'dump': ['existing', 'enoent'],
+}
+
+CONFIG_PUT = []  # list: keys
+CONFIG_DEL = []  # list: keys
+CONFIG_EXISTING = {}  # map: key -> size
+
+
+def run_cmd(cmd, expects=0):
+    full_cmd = ['ceph', 'config-key'] + cmd
+
+    if expects < 0:
+        expects = -expects
+
+    cmdlog = LOG.getChild('run_cmd')
+    cmdlog.debug('{fc}'.format(fc=' '.join(full_cmd)))
+
+    proc = subprocess.Popen(full_cmd,
+                            stdout=subprocess.PIPE,
+                            stderr=subprocess.PIPE)
+
+    stdout = []
+    stderr = []
+    while True:
+        try:
+            out, err = proc.communicate()
+            if out is not None:
+                stdout += out.decode().split('\n')
+                cmdlog.debug('stdout: {s}'.format(s=out))
+            if err is not None:
+                stdout += err.decode().split('\n')
+                cmdlog.debug('stderr: {s}'.format(s=err))
+        except ValueError:
+            ret = proc.wait()
+            break
+
+    if ret != expects:
+        cmdlog.error('cmd > {cmd}'.format(cmd=full_cmd))
+        cmdlog.error("expected return '{expected}' got '{got}'".format(
+            expected=expects, got=ret))
+        cmdlog.error('stdout')
+        for i in stdout:
+            cmdlog.error('{x}'.format(x=i))
+        cmdlog.error('stderr')
+        for i in stderr:
+            cmdlog.error('{x}'.format(x=i))
+
+
+# end run_cmd
+
+def gen_data(size, rnd):
+    chars = string.ascii_letters + string.digits
+    return ''.join(rnd.choice(chars) for _ in range(size))
+
+
+def gen_key(rnd):
+    return gen_data(20, rnd)
+
+
+def gen_tmp_file_path(rnd):
+    file_name = gen_data(20, rnd)
+    file_path = os.path.join('/tmp', 'ceph-test.' + file_name)
+    return file_path
+
+
+def destroy_tmp_file(fpath):
+    if os.path.exists(fpath) and os.path.isfile(fpath):
+        os.unlink(fpath)
+
+
+def write_data_file(data, rnd):
+    file_path = gen_tmp_file_path(rnd)
+    data_file = open(file_path, 'a+')
+    data_file.truncate()
+    data_file.write(data)
+    data_file.close()
+    return file_path
+
+
+# end write_data_file
+
+def choose_random_op(rnd):
+    op = rnd.choice(
+        list(OPS.keys())
+    )
+    sop = rnd.choice(OPS[op])
+    return op, sop
+
+
+def parse_args(args):
+    parser = argparse.ArgumentParser(
+        description="Test the monitor's 'config-key' API",
+    )
+    parser.add_argument(
+        '-v', '--verbose',
+        action='store_true',
+        help='be more verbose',
+    )
+    parser.add_argument(
+        '-s', '--seed',
+        metavar='SEED',
+        help='use SEED instead of generating it in run-time',
+    )
+    parser.add_argument(
+        '-d', '--duration',
+        metavar='SECS',
+        help='run test for SECS seconds (default: 300)',
+    )
+    parser.set_defaults(
+        seed=None,
+        duration=300,
+        verbose=False,
+    )
+    return parser.parse_args(args)
+
+
+def main():
+    args = parse_args(sys.argv[1:])
+
+    verbose = args.verbose
+    if os.environ.get('CEPH_TEST_VERBOSE') is not None:
+        verbose = (os.environ.get('CEPH_TEST_VERBOSE') == '1')
+
+    duration = int(os.environ.get('CEPH_TEST_DURATION', args.duration))
+    seed = os.environ.get('CEPH_TEST_SEED', args.seed)
+    seed = int(time.time()) if seed is None else int(seed)
+
+    rnd = random.Random()
+    rnd.seed(seed)
+
+    loglevel = logging.INFO
+    if verbose:
+        loglevel = logging.DEBUG
+
+    logging.basicConfig(level=loglevel)
+
+    LOG.info('seed: {s}'.format(s=seed))
+
+    start = time.time()
+
+    while (time.time() - start) < duration:
+        (op, sop) = choose_random_op(rnd)
+
+        LOG.info('{o}({s})'.format(o=op, s=sop))
+        op_log = LOG.getChild('{o}({s})'.format(o=op, s=sop))
+
+        if op == 'put':
+            via_file = (rnd.uniform(0, 100) < 50.0)
+
+            expected = 0
+            cmd = ['put']
+            key = None
+
+            if sop == 'existing':
+                if len(CONFIG_EXISTING) == 0:
+                    op_log.debug('no existing keys; continue')
+                    continue
+                key = rnd.choice(CONFIG_PUT)
+                assert key in CONFIG_EXISTING, \
+                    "key '{k_}' not in CONFIG_EXISTING".format(k_=key)
+
+                expected = 0  # the store just overrides the value if the key exists
+            # end if sop == 'existing'
+            elif sop == 'new':
+                for x in range(0, 10):
+                    key = gen_key(rnd)
+                    if key not in CONFIG_EXISTING:
+                        break
+                    key = None
+                if key is None:
+                    op_log.error('unable to generate an unique key -- try again later.')
+                    continue
+
+                assert key not in CONFIG_PUT and key not in CONFIG_EXISTING, \
+                    'key {k} was not supposed to exist!'.format(k=key)
+
+            assert key is not None, \
+                'key must be != None'
+
+            cmd += [key]
+
+            (size, error) = rnd.choice(SIZES)
+            if size > 25:
+                via_file = True
+
+            data = gen_data(size, rnd)
+
+            if error == 0:  # only add if we expect the put to be successful
+                if sop == 'new':
+                    CONFIG_PUT.append(key)
+                CONFIG_EXISTING[key] = size
+            expected = error
+
+            if via_file:
+                data_file = write_data_file(data, rnd)
+                cmd += ['-i', data_file]
+            else:
+                cmd += [data]
+
+            op_log.debug('size: {sz}, via: {v}'.format(
+                sz=size,
+                v='file: {f}'.format(f=data_file) if via_file == True else 'cli')
+            )
+            run_cmd(cmd, expects=expected)
+            if via_file:
+                destroy_tmp_file(data_file)
+            continue
+
+        elif op == 'del':
+            expected = 0
+            cmd = ['del']
+            key = None
+
+            if sop == 'existing':
+                if len(CONFIG_EXISTING) == 0:
+                    op_log.debug('no existing keys; continue')
+                    continue
+                key = rnd.choice(CONFIG_PUT)
+                assert key in CONFIG_EXISTING, \
+                    "key '{k_}' not in CONFIG_EXISTING".format(k_=key)
+
+            if sop == 'enoent':
+                for x in range(0, 10):
+                    key = base64.b64encode(os.urandom(20)).decode()
+                    if key not in CONFIG_EXISTING:
+                        break
+                    key = None
+                if key is None:
+                    op_log.error('unable to generate an unique key -- try again later.')
+                    continue
+                assert key not in CONFIG_PUT and key not in CONFIG_EXISTING, \
+                    'key {k} was not supposed to exist!'.format(k=key)
+                expected = 0  # deleting a non-existent key succeeds
+
+            assert key is not None, \
+                'key must be != None'
+
+            cmd += [key]
+            op_log.debug('key: {k}'.format(k=key))
+            run_cmd(cmd, expects=expected)
+            if sop == 'existing':
+                CONFIG_DEL.append(key)
+                CONFIG_PUT.remove(key)
+                del CONFIG_EXISTING[key]
+            continue
+
+        elif op == 'exists':
+            expected = 0
+            cmd = ['exists']
+            key = None
+
+            if sop == 'existing':
+                if len(CONFIG_EXISTING) == 0:
+                    op_log.debug('no existing keys; continue')
+                    continue
+                key = rnd.choice(CONFIG_PUT)
+                assert key in CONFIG_EXISTING, \
+                    "key '{k_}' not in CONFIG_EXISTING".format(k_=key)
+
+            if sop == 'enoent':
+                for x in range(0, 10):
+                    key = base64.b64encode(os.urandom(20)).decode()
+                    if key not in CONFIG_EXISTING:
+                        break
+                    key = None
+                if key is None:
+                    op_log.error('unable to generate an unique key -- try again later.')
+                    continue
+                assert key not in CONFIG_PUT and key not in CONFIG_EXISTING, \
+                    'key {k} was not supposed to exist!'.format(k=key)
+                expected = -errno.ENOENT
+
+            assert key is not None, \
+                'key must be != None'
+
+            cmd += [key]
+            op_log.debug('key: {k}'.format(k=key))
+            run_cmd(cmd, expects=expected)
+            continue
+
+        elif op == 'get':
+            expected = 0
+            cmd = ['get']
+            key = None
+
+            if sop == 'existing':
+                if len(CONFIG_EXISTING) == 0:
+                    op_log.debug('no existing keys; continue')
+                    continue
+                key = rnd.choice(CONFIG_PUT)
+                assert key in CONFIG_EXISTING, \
+                    "key '{k_}' not in CONFIG_EXISTING".format(k_=key)
+
+            if sop == 'enoent':
+                for x in range(0, 10):
+                    key = base64.b64encode(os.urandom(20)).decode()
+                    if key not in CONFIG_EXISTING:
+                        break
+                    key = None
+                if key is None:
+                    op_log.error('unable to generate an unique key -- try again later.')
+                    continue
+                assert key not in CONFIG_PUT and key not in CONFIG_EXISTING, \
+                    'key {k} was not supposed to exist!'.format(k=key)
+                expected = -errno.ENOENT
+
+            assert key is not None, \
+                'key must be != None'
+
+            file_path = gen_tmp_file_path(rnd)
+            cmd += [key, '-o', file_path]
+            op_log.debug('key: {k}'.format(k=key))
+            run_cmd(cmd, expects=expected)
+            if sop == 'existing':
+                try:
+                    temp_file = open(file_path, 'r+')
+                except IOError as err:
+                    if err.errno == errno.ENOENT:
+                        assert CONFIG_EXISTING[key] == 0, \
+                            "error opening '{fp}': {e}".format(fp=file_path, e=err)
+                        continue
+                    else:
+                        assert False, \
+                            'some error occurred: {e}'.format(e=err)
+                cnt = 0
+                while True:
+                    read_data = temp_file.read()
+                    if read_data == '':
+                        break
+                    cnt += len(read_data)
+                assert cnt == CONFIG_EXISTING[key], \
+                    "wrong size from store for key '{k}': {sz}, expected {es}".format(
+                        k=key, sz=cnt, es=CONFIG_EXISTING[key])
+                destroy_tmp_file(file_path)
+            continue
+
+        elif op == 'list' or op == 'dump':
+            expected = 0
+            cmd = [op]
+            key = None
+
+            if sop == 'existing':
+                if len(CONFIG_EXISTING) == 0:
+                    op_log.debug('no existing keys; continue')
+                    continue
+                key = rnd.choice(CONFIG_PUT)
+                assert key in CONFIG_EXISTING, \
+                    "key '{k_}' not in CONFIG_EXISTING".format(k_=key)
+
+            if sop == 'enoent':
+                for x in range(0, 10):
+                    key = base64.b64encode(os.urandom(20)).decode()
+                    if key not in CONFIG_EXISTING:
+                        break
+                    key = None
+                if key is None:
+                    op_log.error('unable to generate an unique key -- try again later.')
+                    continue
+                assert key not in CONFIG_PUT and key not in CONFIG_EXISTING, \
+                    'key {k} was not supposed to exist!'.format(k=key)
+
+            assert key is not None, \
+                'key must be != None'
+
+            file_path = gen_tmp_file_path(rnd)
+            cmd += ['-o', file_path]
+            op_log.debug('key: {k}'.format(k=key))
+            run_cmd(cmd, expects=expected)
+            try:
+                temp_file = open(file_path, 'r+')
+            except IOError as err:
+                if err.errno == errno.ENOENT:
+                    assert CONFIG_EXISTING[key] == 0, \
+                        "error opening '{fp}': {e}".format(fp=file_path, e=err)
+                    continue
+                else:
+                    assert False, \
+                        'some error occurred: {e}'.format(e=err)
+            cnt = 0
+            try:
+                read_data = json.load(temp_file)
+            except ValueError:
+                temp_file.seek(0)
+                assert False, "{op} output was not valid JSON:\n{filedata}".format(op, temp_file.readlines())
+
+            if sop == 'existing':
+                assert key in read_data, "key '{k}' not found in list/dump output".format(k=key)
+                if op == 'dump':
+                    cnt = len(read_data[key])
+                    assert cnt == CONFIG_EXISTING[key], \
+                        "wrong size from list for key '{k}': {sz}, expected {es}".format(
+                        k=key, sz=cnt, es=CONFIG_EXISTING[key])
+            elif sop == 'enoent':
+                assert key not in read_data, "key '{k}' found in list/dump output".format(k=key)
+            destroy_tmp_file(file_path)
+            continue
+        else:
+            assert False, 'unknown op {o}'.format(o=op)
+
+    # check if all keys in 'CONFIG_PUT' exist and
+    # if all keys on 'CONFIG_DEL' don't.
+    # but first however, remove all keys in CONFIG_PUT that might
+    # be in CONFIG_DEL as well.
+    config_put_set = set(CONFIG_PUT)
+    config_del_set = set(CONFIG_DEL).difference(config_put_set)
+
+    LOG.info('perform sanity checks on store')
+
+    for k in config_put_set:
+        LOG.getChild('check(puts)').debug('key: {k_}'.format(k_=k))
+        run_cmd(['exists', k], expects=0)
+    for k in config_del_set:
+        LOG.getChild('check(dels)').debug('key: {k_}'.format(k_=k))
+        run_cmd(['exists', k], expects=-errno.ENOENT)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/qa/workunits/mon/test_mon_osdmap_prune.sh b/qa/workunits/mon/test_mon_osdmap_prune.sh
new file mode 100755
index 00000000..9cdd7217
--- /dev/null
+++ b/qa/workunits/mon/test_mon_osdmap_prune.sh
@@ -0,0 +1,205 @@
+#!/bin/bash
+
+. $(dirname $0)/../../standalone/ceph-helpers.sh
+
+set -x
+
+function wait_for_osdmap_manifest() {
+
+  local what=${1:-"true"}
+
+  local -a delays=($(get_timeout_delays $TIMEOUT .1))
+  local -i loop=0
+
+  for ((i=0; i < ${#delays[*]}; ++i)); do
+    has_manifest=$(ceph report | jq 'has("osdmap_manifest")')
+    if [[ "$has_manifest" == "$what" ]]; then
+      return 0
+    fi
+
+    sleep ${delays[$i]}
+  done
+
+  echo "osdmap_manifest never outputted on report"
+  ceph report
+  return 1
+}
+
+function wait_for_trim() {
+
+  local -i epoch=$1
+  local -a delays=($(get_timeout_delays $TIMEOUT .1))
+  local -i loop=0
+
+  for ((i=0; i < ${#delays[*]}; ++i)); do
+    fc=$(ceph report | jq '.osdmap_first_committed')
+    if [[ $fc -eq $epoch ]]; then
+      return 0
+    fi
+    sleep ${delays[$i]}
+  done
+
+  echo "never trimmed up to epoch $epoch"
+  ceph report
+  return 1
+}
+
+function test_osdmap() {
+
+  local epoch=$1
+  local ret=0
+
+  tmp_map=$(mktemp)
+  ceph osd getmap $epoch -o $tmp_map || return 1
+  if ! osdmaptool --print $tmp_map | grep "epoch $epoch" ; then
+    echo "ERROR: failed processing osdmap epoch $epoch"
+    ret=1
+  fi
+  rm $tmp_map
+  return $ret
+}
+
+function generate_osdmaps() {
+
+  local -i num=$1
+
+  cmds=( set unset )
+  for ((i=0; i < num; ++i)); do
+    ceph osd ${cmds[$((i%2))]} noup || return 1
+  done
+  return 0
+}
+
+function test_mon_osdmap_prune() {
+
+  create_pool foo 32
+  wait_for_clean || return 1
+
+  ceph config set mon mon_debug_block_osdmap_trim true || return 1
+
+  generate_osdmaps 500 || return 1
+
+  report="$(ceph report)"
+  fc=$(jq '.osdmap_first_committed' <<< $report)
+  lc=$(jq '.osdmap_last_committed' <<< $report)
+
+  [[ $((lc-fc)) -ge 500 ]] || return 1
+
+  wait_for_osdmap_manifest || return 1
+
+  manifest="$(ceph report | jq '.osdmap_manifest')"
+
+  first_pinned=$(jq '.first_pinned' <<< $manifest)
+  last_pinned=$(jq '.last_pinned' <<< $manifest)
+  pinned_maps=( $(jq '.pinned_maps[]' <<< $manifest) )
+
+  # validate pinned maps list
+  [[ $first_pinned -eq ${pinned_maps[0]} ]] || return 1
+  [[ $last_pinned -eq ${pinned_maps[-1]} ]] || return 1
+
+  # validate pinned maps range
+  [[ $first_pinned -lt $last_pinned ]] || return 1
+  [[ $last_pinned -lt $lc ]] || return 1
+  [[ $first_pinned -eq $fc ]] || return 1
+
+  # ensure all the maps are available, and work as expected
+  # this can take a while...
+
+  for ((i=$first_pinned; i <= $last_pinned; ++i)); do
+    test_osdmap $i || return 1
+  done
+
+  # update pinned maps state:
+  #  the monitor may have pruned & pinned additional maps since we last
+  #  assessed state, given it's an iterative process.
+  #
+  manifest="$(ceph report | jq '.osdmap_manifest')"
+  first_pinned=$(jq '.first_pinned' <<< $manifest)
+  last_pinned=$(jq '.last_pinned' <<< $manifest)
+  pinned_maps=( $(jq '.pinned_maps[]' <<< $manifest) )
+
+  # test trimming maps
+  #
+  # we're going to perform the following tests:
+  #
+  #  1. force trim to a pinned map
+  #  2. force trim to a pinned map's previous epoch
+  #  3. trim all maps except the last 200 or so.
+  #
+
+  # 1. force trim to a pinned map
+  #
+  [[ ${#pinned_maps[@]} -gt 10 ]] || return 1
+
+  trim_to=${pinned_maps[1]}
+  ceph config set mon mon_osd_force_trim_to $trim_to
+  ceph config set mon mon_min_osdmap_epochs 100
+  ceph config set mon paxos_service_trim_min 1
+  ceph config set mon mon_debug_block_osdmap_trim false
+
+  # generate an epoch so we get to trim maps
+  ceph osd set noup
+  ceph osd unset noup
+
+  wait_for_trim $trim_to || return 1
+
+  report="$(ceph report)"
+  fc=$(jq '.osdmap_first_committed' <<< $report)
+  [[ $fc -eq $trim_to ]] || return 1
+
+  old_first_pinned=$first_pinned
+  old_last_pinned=$last_pinned
+  first_pinned=$(jq '.osdmap_manifest.first_pinned' <<< $report)
+  last_pinned=$(jq '.osdmap_manifest.last_pinned' <<< $report)
+  [[ $first_pinned -eq $trim_to ]] || return 1
+  [[ $first_pinned -gt $old_first_pinned ]] || return 1
+  [[ $last_pinned -gt $old_first_pinned ]] || return 1
+
+  test_osdmap $trim_to || return 1
+  test_osdmap $(( trim_to+1 )) || return 1
+
+  pinned_maps=( $(jq '.osdmap_manifest.pinned_maps[]' <<< $report) )
+
+  # 2. force trim to a pinned map's previous epoch
+  #
+  [[ ${#pinned_maps[@]} -gt 2 ]] || return 1
+  trim_to=$(( ${pinned_maps[1]} - 1))
+  ceph config set mon mon_osd_force_trim_to $trim_to
+
+  # generate an epoch so we get to trim maps
+  ceph osd set noup
+  ceph osd unset noup
+
+  wait_for_trim $trim_to || return 1
+
+  report="$(ceph report)"
+  fc=$(jq '.osdmap_first_committed' <<< $report)
+  [[ $fc -eq $trim_to ]] || return 1
+
+  old_first_pinned=$first_pinned
+  old_last_pinned=$last_pinned
+  first_pinned=$(jq '.osdmap_manifest.first_pinned' <<< $report)
+  last_pinned=$(jq '.osdmap_manifest.last_pinned' <<< $report)
+  pinned_maps=( $(jq '.osdmap_manifest.pinned_maps[]' <<< $report) )
+  [[ $first_pinned -eq $trim_to ]] || return 1
+  [[ ${pinned_maps[1]} -eq $(( trim_to+1)) ]] || return 1
+
+  test_osdmap $first_pinned || return 1
+  test_osdmap $(( first_pinned + 1 )) || return 1
+
+  # 3. trim everything
+  #
+  ceph config set mon mon_osd_force_trim_to 0
+
+  # generate an epoch so we get to trim maps
+  ceph osd set noup
+  ceph osd unset noup
+
+  wait_for_osdmap_manifest "false" || return 1
+
+  return 0
+}
+
+test_mon_osdmap_prune || exit 1
+
+echo "OK"