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