scrub/xfs_scrub_all.in

   1 #!/usr/bin/python3
   2
   3 # Run online scrubbers in parallel, but avoid thrashing.
   4 #
   5 # Copyright (C) 2018 Oracle.  All rights reserved.
   6 #
   7 # Author: Darrick J. Wong <darrick.wong@oracle.com>
   8 #
   9 # This program is free software; you can redistribute it and/or
  10 # modify it under the terms of the GNU General Public License
  11 # as published by the Free Software Foundation; either version 2
  12 # of the License, or (at your option) any later version.
  13 #
  14 # This program is distributed in the hope that it would be useful,
  15 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  16 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  17 # GNU General Public License for more details.
  18 #
  19 # You should have received a copy of the GNU General Public License
  20 # along with this program; if not, write the Free Software Foundation,
  21 # Inc.,  51 Franklin St, Fifth Floor, Boston, MA  02110-1301, USA.
  22
  23 import subprocess
  24 import json
  25 import threading
  26 import time
  27 import sys
  28 import os
  29 import argparse
  30
  31 retcode = 0
  32 terminate = False
  33
  34 def DEVNULL():
  35         '''Return /dev/null in subprocess writable format.'''
  36         try:
  37                 from subprocess import DEVNULL
  38                 return DEVNULL
  39         except ImportError:
  40                 return open(os.devnull, 'wb')
  41
  42 def find_mounts():
  43         '''Map mountpoints to physical disks.'''
  44
  45         fs = {}
  46         cmd=['lsblk', '-o', 'KNAME,TYPE,FSTYPE,MOUNTPOINT', '-J']
  47         result = subprocess.Popen(cmd, stdout=subprocess.PIPE)
  48         result.wait()
  49         if result.returncode != 0:
  50                 return fs
  51         sarray = [x.decode('utf-8') for x in result.stdout.readlines()]
  52         output = ' '.join(sarray)
  53         bdevdata = json.loads(output)
  54         # The lsblk output had better be in disks-then-partitions order
  55         for bdev in bdevdata['blockdevices']:
  56                 if bdev['type'] in ('disk', 'loop'):
  57                         lastdisk = bdev['kname']
  58                 if bdev['fstype'] == 'xfs':
  59                         mnt = bdev['mountpoint']
  60                         if mnt is None:
  61                                 continue
  62                         if mnt in fs:
  63                                 fs[mnt].add(lastdisk)
  64                         else:
  65                                 fs[mnt] = set([lastdisk])
  66         return fs
  67
  68 def kill_systemd(unit, proc):
  69         '''Kill systemd unit.'''
  70         proc.terminate()
  71         cmd=['systemctl', 'stop', unit]
  72         x = subprocess.Popen(cmd)
  73         x.wait()
  74
  75 def run_killable(cmd, stdout, killfuncs, kill_fn):
  76         '''Run a killable program.  Returns program retcode or -1 if we can't start it.'''
  77         try:
  78                 proc = subprocess.Popen(cmd, stdout = stdout)
  79                 real_kill_fn = lambda: kill_fn(proc)
  80                 killfuncs.add(real_kill_fn)
  81                 proc.wait()
  82                 try:
  83                         killfuncs.remove(real_kill_fn)
  84                 except:
  85                         pass
  86                 return proc.returncode
  87         except:
  88                 return -1
  89
  90 # systemd doesn't like unit instance names with slashes in them, so it
  91 # replaces them with dashes when it invokes the service.  However, it's not
  92 # smart enough to convert the dashes to something else, so when it unescapes
  93 # the instance name to feed to xfs_scrub, it turns all dashes into slashes.
  94 # "/moo-cow" becomes "-moo-cow" becomes "/moo/cow", which is wrong.  systemd
  95 # actually /can/ escape the dashes correctly if it is told that this is a path
  96 # (and not a unit name), but it didn't do this prior to January 2017, so fix
  97 # this for them.
  98 def systemd_escape(path):
  99         '''Escape a path to avoid mangled systemd mangling.'''
 100
 101         if '-' not in path:
 102                 return path
 103         cmd = ['systemd-escape', '--path', path]
 104         try:
 105                 proc = subprocess.Popen(cmd, stdout = subprocess.PIPE)
 106                 proc.wait()
 107                 for line in proc.stdout:
 108                         return '-' + line.decode(sys.stdout.encoding).strip()
 109         except:
 110                 return path
 111
 112 def run_scrub(mnt, cond, running_devs, mntdevs, killfuncs):
 113         '''Run a scrub process.'''
 114         global retcode, terminate
 115
 116         print("Scrubbing %s..." % mnt)
 117         sys.stdout.flush()
 118
 119         try:
 120                 if terminate:
 121                         return
 122
 123                 # Try it the systemd way
 124                 cmd=['systemctl', 'start', 'xfs_scrub@%s' % systemd_escape(mnt)]
 125                 ret = run_killable(cmd, DEVNULL(), killfuncs, \
 126                                 lambda proc: kill_systemd('xfs_scrub@%s' % mnt, proc))
 127                 if ret == 0 or ret == 1:
 128                         print("Scrubbing %s done, (err=%d)" % (mnt, ret))
 129                         sys.stdout.flush()
 130                         retcode |= ret
 131                         return
 132
 133                 if terminate:
 134                         return
 135
 136                 # Invoke xfs_scrub manually
 137                 cmd=['@sbindir@/xfs_scrub', '@scrub_args@', mnt]
 138                 ret = run_killable(cmd, None, killfuncs, \
 139                                 lambda proc: proc.terminate())
 140                 if ret >= 0:
 141                         print("Scrubbing %s done, (err=%d)" % (mnt, ret))
 142                         sys.stdout.flush()
 143                         retcode |= ret
 144                         return
 145
 146                 if terminate:
 147                         return
 148
 149                 print("Unable to start scrub tool.")
 150                 sys.stdout.flush()
 151         finally:
 152                 running_devs -= mntdevs
 153                 cond.acquire()
 154                 cond.notify()
 155                 cond.release()
 156
 157 def main():
 158         '''Find mounts, schedule scrub runs.'''
 159         def thr(mnt, devs):
 160                 a = (mnt, cond, running_devs, devs, killfuncs)
 161                 thr = threading.Thread(target = run_scrub, args = a)
 162                 thr.start()
 163         global retcode, terminate
 164
 165         parser = argparse.ArgumentParser( \
 166                         description = "Scrub all mounted XFS filesystems.")
 167         parser.add_argument("-V", help = "Report version and exit.", \
 168                         action = "store_true")
 169         args = parser.parse_args()
 170
 171         if args.V:
 172                 print("xfs_scrub_all version @pkg_version@")
 173                 sys.exit(0)
 174
 175         fs = find_mounts()
 176
 177         # Tail the journal if we ourselves aren't a service...
 178         journalthread = None
 179         if 'SERVICE_MODE' not in os.environ:
 180                 try:
 181                         cmd=['journalctl', '--no-pager', '-q', '-S', 'now', \
 182                                         '-f', '-u', 'xfs_scrub@*', '-o', \
 183                                         'cat']
 184                         journalthread = subprocess.Popen(cmd)
 185                 except:
 186                         pass
 187
 188         # Schedule scrub jobs...
 189         running_devs = set()
 190         killfuncs = set()
 191         cond = threading.Condition()
 192         while len(fs) > 0:
 193                 if len(running_devs) == 0:
 194                         mnt, devs = fs.popitem()
 195                         running_devs.update(devs)
 196                         thr(mnt, devs)
 197                 poppers = set()
 198                 for mnt in fs:
 199                         devs = fs[mnt]
 200                         can_run = True
 201                         for dev in devs:
 202                                 if dev in running_devs:
 203                                         can_run = False
 204                                         break
 205                         if can_run:
 206                                 running_devs.update(devs)
 207                                 poppers.add(mnt)
 208                                 thr(mnt, devs)
 209                 for p in poppers:
 210                         fs.pop(p)
 211                 cond.acquire()
 212                 try:
 213                         cond.wait()
 214                 except KeyboardInterrupt:
 215                         terminate = True
 216                         print("Terminating...")
 217                         sys.stdout.flush()
 218                         while len(killfuncs) > 0:
 219                                 fn = killfuncs.pop()
 220                                 fn()
 221                         fs = []
 222                 cond.release()
 223
 224         if journalthread is not None:
 225                 journalthread.terminate()
 226
 227         # See the service mode comments in xfs_scrub.c for why we do this.
 228         if 'SERVICE_MODE' in os.environ:
 229                 time.sleep(2)
 230                 if retcode != 0:
 231                         retcode = 1
 232
 233         sys.exit(retcode)
 234
 235 if __name__ == '__main__':
 236         main()