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