diff --git a/tools/upmap/upmap-remapped.py b/tools/upmap/upmap-remapped.py index e40f4d9..69db84a 100755 --- a/tools/upmap/upmap-remapped.py +++ b/tools/upmap/upmap-remapped.py @@ -38,92 +38,129 @@ # Hacked by: Dan van der Ster -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()