tools/upmap: upmap-remapped.py correctness and error handling fixes

tools/upmap/upmap-remapped.py

@@ -38,92 +38,129 @@
 # Hacked by: Dan van der Ster <daniel.vanderster@cern.ch>
 
 
-import json, subprocess, sys
+import argparse, atexit, json, subprocess, sys
+
+# How long to wait for a mon command.  'pg ls' and 'osd dump' can take a while
+# on a large cluster with many remapped pgs.
+MON_TIMEOUT = 300
 
 def get_command_output(command):
   result = subprocess.run(command, capture_output=True, universal_newlines=True, check=True, shell=True)
   return result.stdout
 
+def eprint(*args, **kwargs):
+  print(*args, file=sys.stderr, **kwargs)
+
+parser = argparse.ArgumentParser(
+  description='Print the ceph commands which make every remapped pg '
+              'active+clean again.  Pipe the output into sh to run them.')
+parser.add_argument('--ignore-backfilling', action='store_true',
+                    help='leave the pgs which are already backfilling alone, '
+                         'instead of interrupting them')
+options = parser.parse_args()
+if options.ignore_backfilling:
+  eprint('All actively backfilling PGs will be ignored.')
+
 try:
   import rados
   cluster = rados.Rados(conffile='/etc/ceph/ceph.conf')
   cluster.connect()
-except:
+except Exception:
   use_shell = True
 else:
   use_shell = False
+  # every exit from here on should close the connection, not just the last one
+  atexit.register(cluster.shutdown)
 
-def eprint(*args, **kwargs):
-  print(*args, file=sys.stderr, **kwargs)
+def get_cluster_output(shell_command, mon_command):
+  """Run a command through librados if it is available, else through the shell,
+  and return its output.  Exits if the command fails."""
+  try:
+    if use_shell:
+      return get_command_output(shell_command)
+    ret, output, errs = cluster.mon_command(json.dumps(mon_command), b'',
+                                            timeout=MON_TIMEOUT)
+  except Exception as e:
+    eprint('Error running "%s": %s' % (shell_command, e))
+    sys.exit(1)
+  if ret != 0:
+    eprint('Error running "%s": %s'
+           % (shell_command, errs.strip() or 'returned %d' % ret))
+    sys.exit(1)
+  return output.decode('utf-8').strip()
 
 try:
-  if use_shell:
-    OSDS = json.loads(get_command_output('ceph osd ls -f json | jq -r .'))
-    DF = json.loads(get_command_output('ceph osd df -f json | jq -r .nodes'))
-  else:
-    cmd = {"prefix": "osd ls", "format": "json"}
-    ret, output, errs = cluster.mon_command(json.dumps(cmd), b'', timeout=5)
-    output = output.decode('utf-8').strip()
-    OSDS = json.loads(output)
-    cmd = {"prefix": "osd df", "format": "json"}
-    ret, output, errs = cluster.mon_command(json.dumps(cmd), b'', timeout=5)
-    output = output.decode('utf-8').strip()
-    DF = json.loads(output)['nodes']
+  OSDS = set(json.loads(get_cluster_output('ceph osd ls -f json',
+                                           {"prefix": "osd ls", "format": "json"})))
+  DF = json.loads(get_cluster_output('ceph osd df -f json',
+                                     {"prefix": "osd df", "format": "json"}))['nodes']
 except ValueError:
   eprint('Error loading OSD IDs')
   sys.exit(1)
 
-ignore_backfilling = False
-for arg in sys.argv[1:]:
-  if arg == "--ignore-backfilling":
-    eprint ("All actively backfilling PGs will be ignored.")
-    ignore_backfilling = True
+# the weight each osd effectively has, indexed by osd id: gen_upmap() asks about
+# this for every shard of every remapped pg
+WEIGHT = dict((o['id'], o['crush_weight'] * o['reweight']) for o in DF)
 
 def crush_weight(id):
-  for o in DF:
-    if o['id'] == id:
-      return o['crush_weight'] * o['reweight']
-  return 0
+  return WEIGHT.get(id, 0)
 
 def gen_upmap(up, acting, replicated=False):
-  assert(len(up) == len(acting))
+  # a pg which is degraded as well as remapped can report an acting set of a
+  # different length, and there is nothing useful to do with those
+  if len(up) != len(acting):
+    return []
+
+  # On replicated pools only the set of osds matters, so vacate the osds which do
+  # not belong in the pg and fill it with the ones which are missing from it.
+  # This never maps onto an osd which is already in the up set, which the mon
+  # would ignore.
+  # e.g. ceph osd pg-upmap-items 4.5fd 603 383 499 804
+  if replicated:
+    sources = [u for u in up if u not in acting and u in OSDS]
+    dests = [a for a in acting if a not in up and crush_weight(a) > 0]
+    return list(zip(sources, dests))
 
-  # Create mappings needed to make the PG clean
+  # On erasure-coded pools every position in the up set matters, so the mappings
+  # have to be positional.  Only keep the ones we are allowed to make.
   mappings = [(u, a) for u, a in zip(up, acting) if u != a and u in OSDS and crush_weight(a) > 0]
 
-  # Remove indirect mappings on replicated pools
-  # e.g. ceph osd pg-upmap-items 4.5fd 603 383 499 804 804 530 &
-  if replicated:
-    p = list(mappings)
-    u = set([x[0] for x in p])
-    a = set([x[1] for x in p])
-    mappings = list(zip(u-a, a-u))
+  # Dropping a mapping above leaves its osd in the up set, and mapping onto an
+  # osd which is staying in the up set asks for the same osd twice, which the mon
+  # ignores.  Drop those mappings too, repeating until nothing changes.
+  while True:
+    staying = set(up) - set(u for u, a in mappings)
+    keep = [(u, a) for u, a in mappings if a not in staying]
+    if len(keep) == len(mappings):
+      break
+    mappings = keep
+
   # Order the mappings on erasure-coded pools so that data is moved off an osd
   # before it is moved on to it.
   # e.g. ceph osd pg-upmap-items 15.c9 714 803 929 714
-  else:
-    # Handle the situation where the src and dst of one mapping matches the dst
-    # and src of another.  Example: (314, 272) & (272, 314)
-    for (x, y) in mappings:
-      if (y, x) in mappings:
-        mappings.remove((x, y))
-        mappings.remove((y, x))
-
-    # Do multiple passes of a modified bubble sort to order the mappings so that
-    # data is moved off an OSD before it is moved on to it.  Stop when no
-    # mappings are swapped.
-    while True:
-      swapped = False
-      for i in range(len(mappings)-1):
-        for j in range(i+1, len(mappings)):
-          if mappings[j][0] == mappings[i][1] and mappings[j][1] != mappings[i][0]:
-            mappings[i], mappings[j] = mappings[j], mappings[i]
-            swapped = True
-
-      if not swapped:
-        break
-
-  return mappings
+  # Each osd is used at most once as a source and once as a destination, so the
+  # mappings form chains and cycles.  Emit each chain in order.  A cycle, such as
+  # (314, 272) & (272, 314) or 1 -> 2 -> 3 -> 1, has no valid order, so leave
+  # those mappings out and let the pg stay remapped.
+  by_source = dict((u, (u, a)) for u, a in mappings)
+  ordered = []
+  placed = set()
+  for m in mappings:
+    if m in placed:
+      continue
+    # walk back over the mappings which have to be done before this one
+    chain = []
+    n = m
+    while n is not None and n not in placed and n not in chain:
+      chain.append(n)
+      n = by_source.get(n[1])
+    placed.update(chain)
+    if n in chain:
+      continue
+    chain.reverse()
+    ordered.extend(chain)
+
+  return ordered
 
 def upmap_pg_items(pgid, mapping):
   if len(mapping):
@@ -140,12 +177,8 @@ def rm_upmap_pg_items(pgid):
 
 # discover remapped pgs
 try:
-  if use_shell:
-    remapped_json = get_command_output('ceph pg ls remapped -f json | jq -r .')
-  else:
-    cmd = {"prefix": "pg ls", "states": ["remapped"], "format": "json"}
-    ret, output, err = cluster.mon_command(json.dumps(cmd), b'', timeout=5)
-    remapped_json = output.decode('utf-8').strip()
+  remapped_json = get_cluster_output('ceph pg ls remapped -f json',
+                                     {"prefix": "pg ls", "states": ["remapped"], "format": "json"})
   try:
     remapped = json.loads(remapped_json)['pg_stats']
   except KeyError:
@@ -157,12 +190,8 @@ def rm_upmap_pg_items(pgid):
 
 # discover existing upmaps
 try:
-  if use_shell:
-    osd_dump_json = get_command_output('ceph osd dump -f json | jq -r .')
-  else:
-    cmd = {"prefix": "osd dump", "format": "json"}
-    ret, output, errs = cluster.mon_command(json.dumps(cmd), b'', timeout=5)
-    osd_dump_json = output.decode('utf-8').strip()
+  osd_dump_json = get_cluster_output('ceph osd dump -f json',
+                                     {"prefix": "osd dump", "format": "json"})
   upmaps = json.loads(osd_dump_json)['pg_upmap_items']
 except ValueError:
   eprint('Error loading existing upmaps')
@@ -171,25 +200,18 @@ def rm_upmap_pg_items(pgid):
 # discover pools replicated or erasure
 pool_type = {}
 try:
-  if use_shell:
-    osd_pool_ls_detail = get_command_output('ceph osd pool ls detail')
-  else:
-    cmd = {"prefix": "osd pool ls", "detail": "detail", "format": "plain"}
-    ret, output, errs = cluster.mon_command(json.dumps(cmd), b'', timeout=5)
-    osd_pool_ls_detail = output.decode('utf-8').strip()
+  osd_pool_ls_detail = get_cluster_output('ceph osd pool ls detail',
+                                          {"prefix": "osd pool ls", "detail": "detail", "format": "plain"})
   for line in osd_pool_ls_detail.split('\n'):
-    if 'pool' in line:
+    if line.startswith('pool '):
       x = line.split(' ')
       pool_type[x[1]] = x[3]
-except:
+except IndexError:
   eprint('Error parsing pool types')
   sys.exit(1)
 
 # discover if each pg is already upmapped
-has_upmap = {}
-for pg in upmaps:
-  pgid = str(pg['pgid'])
-  has_upmap[pgid] = True
+has_upmap = set(str(pg['pgid']) for pg in upmaps)
 
 # handle each remapped pg
 print(r'while ceph status | grep -q "peering\|activating\|laggy"; do sleep 2; done')
@@ -199,40 +221,28 @@ def rm_upmap_pg_items(pgid):
     print(r'wait; sleep 4; while ceph status | grep -q "peering\|activating\|laggy"; do sleep 2; done')
     num = 0
 
-  if ignore_backfilling:
-    if "backfilling" in pg['state']:
-      continue
+  if options.ignore_backfilling and "backfilling" in pg['state']:
+    continue
 
   pgid = pg['pgid']
 
-  try:
-    if has_upmap[pgid]:
-      rm_upmap_pg_items(pgid)
-      num += 1
-      continue
-  except KeyError:
-    pass
+  if pgid in has_upmap:
+    rm_upmap_pg_items(pgid)
+    num += 1
+    continue
 
-  up = pg['up']
-  acting = pg['acting']
   pool = pgid.split('.')[0]
-  if pool_type[pool] == 'replicated':
-    try:
-      pairs = gen_upmap(up, acting, replicated=True)
-    except:
-      continue
-  elif pool_type[pool] == 'erasure':
-    try:
-      pairs = gen_upmap(up, acting)
-    except:
-      continue
-  else:
+  if pool not in pool_type:
+    # the pool was deleted between reading the pgs and reading the pools
+    eprint('Skipping pg %s of unknown pool %s' % (pgid, pool))
+    continue
+  if pool_type[pool] not in ('replicated', 'erasure'):
     eprint('Unknown pool type for %s' % pool)
     sys.exit(1)
+
+  pairs = gen_upmap(pg['up'], pg['acting'],
+                    replicated=(pool_type[pool] == 'replicated'))
   upmap_pg_items(pgid, pairs)
   num += 1
 
 print(r'wait; sleep 4; while ceph status | grep -q "peering\|activating\|laggy"; do sleep 2; done')
-
-if not use_shell:
-  cluster.shutdown()