htsworkflow/automation/spoolwatcher.py

   1 #!/usr/bin/env python
   2 import logging
   3 import os
   4 import re
   5 import shlex
   6 import sys
   7 import time
   8
   9 from htsworkflow.util import mount
  10
  11 # this uses pyinotify
  12 import pyinotify
  13 from pyinotify import EventsCodes
  14
  15 from benderjab import rpc
  16
  17 def get_top_dir(root, path):
  18     """
  19     Return the directory in path that is a subdirectory of root.
  20     e.g.
  21
  22     >>> print get_top_dir('/a/b/c', '/a/b/c/d/e/f')
  23     d
  24     >>> print get_top_dir('/a/b/c/', '/a/b/c/d/e/f')
  25     d
  26     >>> print get_top_dir('/a/b/c', '/g/e/f')
  27     None
  28     >>> print get_top_dir('/a/b/c', '/a/b/c')
  29     <BLANKLINE>
  30     """
  31     if path.startswith(root):
  32         subpath = path[len(root):]
  33         if subpath.startswith('/'):
  34             subpath = subpath[1:]
  35         return subpath.split(os.path.sep)[0]
  36     else:
  37         return None
  38
  39 class WatcherEvents(object):
  40     # two events need to be tracked
  41     # one to send startCopy
  42     # one to send OMG its broken
  43     # OMG its broken needs to stop when we've seen enough
  44     #  cycles
  45     # this should be per runfolder.
  46     # read the xml files
  47     def __init__(self):
  48         pass
  49
  50
  51 class Handler(pyinotify.ProcessEvent):
  52     def __init__(self, watchmanager, bot, ipar=False):
  53         """
  54         ipar flag indicates we should wait for ipar to finish, instead of
  55              just the run finishing
  56         """
  57         self.last_event = {}
  58         self.watchmanager = watchmanager
  59         self.bot = bot
  60         self.ipar_mode = ipar
  61         if self.ipar_mode:
  62             self.last_file = 'IPAR_Netcopy_Complete.txt'.lower()
  63         else:
  64             self.last_file = "run.completed".lower()
  65
  66     def process_IN_CREATE(self, event):
  67         for wdd in self.bot.wdds:
  68             for watch_path in self.bot.watchdirs:
  69                 if event.path.startswith(watch_path):
  70                     target = get_top_dir(watch_path, event.path)
  71                     self.last_event.setdefault(watch_path, {})[target] = time.time()
  72
  73                     msg = "Create: %s %s %s" % (event.path, event.name, target)
  74
  75                     if event.name.lower() == self.last_file:
  76                         try:
  77                             self.bot.sequencingFinished(event.path)
  78                         except IOError, e:
  79                             logging.error("Couldn't send sequencingFinished")
  80                     logging.debug(msg)
  81
  82     def process_IN_DELETE(self, event):
  83         logging.debug("Remove: %s" %  os.path.join(event.path, event.name))
  84         pass
  85
  86     def process_IN_UNMOUNT(self, event):
  87         pathname = os.path.join(event.path, event.name)
  88         logging.debug("IN_UNMOUNT: %s" % (pathname,))
  89         self.bot.unmount_watch(event.path)
  90
  91 class SpoolWatcher(rpc.XmlRpcBot):
  92     """
  93     Watch a directory and send a message when another process is done writing.
  94
  95     This monitors a directory tree using inotify (linux specific) and
  96     after some files having been written will send a message after <timeout>
  97     seconds of no file writing.
  98
  99     (Basically when the solexa machine finishes dumping a round of data
 100     this'll hopefully send out a message saying hey look theres data available
 101
 102     """
 103     # these params need to be in the config file
 104     # I wonder where I should put the documentation
 105     #:Parameters:
 106     #    `watchdirs` - list of directories to monitor for modifications
 107     #    `profile` - specify which .htsworkflow profile to use
 108     #    `write_timeout` - how many seconds to wait for writes to finish to
 109     #                      the spool
 110     #    `notify_timeout` - how often to timeout from notify
 111
 112     def __init__(self, section=None, configfile=None):
 113         #if configfile is None:
 114         #    self.configfile = "~/.htsworkflow"
 115         super(SpoolWatcher, self).__init__(section, configfile)
 116
 117         self.cfg['watchdirs'] = None
 118         self.cfg['write_timeout'] = 10
 119         self.cfg['notify_users'] = None
 120         self.cfg['notify_runner'] = None
 121         self.cfg['wait_for_ipar'] = 0
 122
 123         self.watchdirs = []
 124         self.watchdir_url_map = {}
 125         self.notify_timeout = 0.001
 126
 127         self.wm = None
 128         self.notify_users = None
 129         self.notify_runner = None
 130         self.wdds = []
 131
 132         # keep track if the specified mount point is currently mounted
 133         self.mounted_points = {}
 134         # keep track of which mount points tie to which watch directories
 135         # so maybe we can remount them.
 136         self.mounts_to_watches = {}
 137
 138         self.eventTasks.append(self.process_notify)
 139
 140     def read_config(self, section=None, configfile=None):
 141         # Don't give in to the temptation to use logging functions here,
 142         # need to wait until after we detach in start
 143         super(SpoolWatcher, self).read_config(section, configfile)
 144
 145         self.watchdirs = shlex.split(self._check_required_option('watchdirs'))
 146         # see if there's an alternate url that should be used for the watchdir
 147         for watchdir in self.watchdirs:
 148             self.watchdir_url_map[watchdir] = self.cfg.get(watchdir, watchdir)
 149
 150         self.write_timeout = int(self.cfg['write_timeout'])
 151         self.wait_for_ipar = int(self.cfg['wait_for_ipar'])
 152
 153         self.notify_users = self._parse_user_list(self.cfg['notify_users'])
 154         try:
 155           self.notify_runner = \
 156              self._parse_user_list(self.cfg['notify_runner'],
 157                                    require_resource=True)
 158         except bot.JIDMissingResource:
 159             msg = 'need a full jabber ID + resource for xml-rpc destinations'
 160             raise bot.JIDMissingResource(msg)
 161
 162         self.handler = None
 163         self.notifier = None
 164
 165     def add_watch(self, watchdirs=None):
 166         """
 167         start watching watchdir or self.watchdir
 168         we're currently limited to watching one directory tree.
 169         """
 170         # create the watch managers if we need them
 171         if self.wm is None:
 172             self.wm = pyinotify.WatchManager()
 173             self.handler = Handler(self.wm, self, self.wait_for_ipar)
 174             self.notifier = pyinotify.Notifier(self.wm, self.handler)
 175
 176         # the one tree limit is mostly because self.wdd is a single item
 177         # but managing it as a list might be a bit more annoying
 178         if watchdirs is None:
 179             watchdirs = self.watchdirs
 180
 181         mask = EventsCodes.IN_CREATE | EventsCodes.IN_UNMOUNT
 182         # rec traverses the tree and adds all the directories that are there
 183         # at the start.
 184         # auto_add will add in new directories as they are created
 185         for w in watchdirs:
 186             mount_location = mount.find_mount_point_for(w)
 187             self.mounted_points[mount_location] = True
 188             mounts = self.mounts_to_watches.get(mount_location, [])
 189             if w not in mounts:
 190                 mounts.append(w)
 191                 self.mounts_to_watches[mount_location] = mounts
 192
 193             logging.info(u"Watching:"+unicode(w))
 194             self.wdds.append(self.wm.add_watch(w, mask, rec=True, auto_add=True))
 195
 196     def unmount_watch(self, event_path):
 197         # remove backwards so we don't get weirdness from
 198         # the list getting shorter
 199         for i in range(len(self.wdds),0, -1):
 200             wdd = self.wdds[i]
 201             logging.info(u'unmounting: '+unicode(wdd.items()))
 202             self.wm.rm_watch(wdd.values())
 203             del self.wdds[i]
 204         self.mounted = False
 205
 206     def make_copy_url(self, watchdir, list_event_dir):
 207         root_copy_url = self.watchdir_url_map[watchdir]
 208         if root_copy_url[-1] != '/':
 209             root_copy_url += '/'
 210         copy_url = root_copy_url + list_event_dir
 211         logging.debug('Copy url: %s' % (copy_url,))
 212         return copy_url
 213
 214     def process_notify(self, *args):
 215         if self.notifier is None:
 216             # nothing to do yet
 217             return
 218         # process the queue of events as explained above
 219         self.notifier.process_events()
 220         #check events waits timeout
 221         if self.notifier.check_events(self.notify_timeout):
 222             # read notified events and enqeue them
 223             self.notifier.read_events()
 224             # should we do something?
 225         # has something happened?
 226         for watchdir, last_events in self.handler.last_event.items():
 227             for last_event_dir, last_event_time in last_events.items():
 228                 time_delta = time.time() - last_event_time
 229                 if time_delta > self.write_timeout:
 230                     copy_url = self.make_copy_url(watchdir, last_event_dir)
 231                     self.startCopy(copy_url)
 232                     self.handler.last_event[watchdir] = {}
 233         # handle unmounted filesystems
 234         for mount_point, was_mounted in self.mounted_points.items():
 235             if not was_mounted and mount.is_mounted(mount_point):
 236                 # we've been remounted. Huzzah!
 237                 # restart the watch
 238                 for watch in self.mounts_to_watches[mount_point]:
 239                     self.add_watch(watch)
 240                     logging.info(
 241                         "%s was remounted, restarting watch" % \
 242                             (mount_point)
 243                     )
 244                 self.mounted_points[mount_point] = True
 245
 246     def _parser(self, msg, who):
 247         """
 248         Parse xmpp chat messages
 249         """
 250         help = u"I can send [copy] message, or squencer [finished]"
 251         if re.match(u"help", msg):
 252             reply = help
 253         elif re.match("copy", msg):
 254             self.startCopy(msg)
 255             reply = u"sent copy message"
 256         elif re.match(u"finished", msg):
 257             words = msg.split()
 258             if len(words) == 2:
 259                 self.sequencingFinished(words[1])
 260                 reply = u"sending sequencing finished for %s" % (words[1])
 261             else:
 262                 reply = u"need runfolder name"
 263         else:
 264             reply = u"I didn't understand '%s'" %(msg)
 265         return reply
 266
 267     def run(self):
 268         """
 269         Start application
 270         """
 271         # we have to configure pyinotify after BenderJab.start is called
 272         # as weird things happen to pyinotify if the stdio is closed
 273         # after it's initialized.
 274         self.add_watch()
 275         super(SpoolWatcher, self).run()
 276
 277     def stop(self):
 278         """
 279         shutdown application
 280         """
 281         # destroy the inotify's instance on this interrupt (stop monitoring)
 282         if self.notifier is not None:
 283             self.notifier.stop()
 284         super(SpoolWatcher, self).stop()
 285
 286     def startCopy(self, copy_url=None):
 287         logging.debug("writes seem to have stopped")
 288         if self.notify_runner is not None:
 289             for r in self.notify_runner:
 290                 self.rpc_send(r, tuple([copy_url]), 'startCopy')
 291         if self.notify_users is not None:
 292             for u in self.notify_users:
 293                 self.send(u, 'startCopy %s.' % (copy_urls,))
 294
 295     def sequencingFinished(self, run_dir):
 296         # need to strip off self.watchdirs from rundir I suspect.
 297         logging.info("run.completed in " + str(run_dir))
 298         for watch in self.watchdirs:
 299             if not run_dir.startswith(watch):
 300                 continue
 301             if watch[-1] != os.path.sep:
 302                 watch += os.path.sep
 303             stripped_run_dir = re.sub(watch, "", run_dir)
 304             logging.debug("stripped to " + stripped_run_dir)
 305             if self.notify_users is not None:
 306                 for u in self.notify_users:
 307                     self.send(u, 'Sequencing run %s finished' % \
 308                               (stripped_run_dir))
 309             if self.notify_runner is not None:
 310                 for r in self.notify_runner:
 311                     self.rpc_send(r, (stripped_run_dir,), 'sequencingFinished')
 312
 313 def main(args=None):
 314     bot = SpoolWatcher()
 315     return bot.main(args)
 316
 317 if __name__ == "__main__":
 318     ret = main(sys.argv[1:])
 319     #sys.exit(ret)
 320
 321 # TODO:
 322 # send messages to copier specifying which mount to copy