Implement toolbox monitoring.

- Add monitor tag on toolbox root elements to force Galaxy to watch the toolbox for changes.
 - Refactoring and test improvements for existing tool monitoring code.
 - Implement toolbox shutdown process to ensure different watchers don't compete with each other.
 - Set the default tool conf to be monitored.

config/tool_conf.xml.sample

@@ -1,5 +1,5 @@
 <?xml version='1.0' encoding='utf-8'?>
-<toolbox>
+<toolbox monitor="true">
   <section id="getext" name="Get Data">
     <tool file="data_source/upload.xml" />
     <tool file="data_source/ucsc_tablebrowser.xml" />

lib/galaxy/config.py

@@ -13,6 +13,7 @@ import socket
 import string
 import sys
 import tempfile
+import threading
 from datetime import timedelta
 from galaxy.exceptions import ConfigurationError
 from galaxy.util import listify
@@ -752,8 +753,12 @@ class ConfiguresGalaxyMixin:
             tool_configs.append( self.config.migrated_tools_config )
 
         from galaxy import tools
-        self.toolbox = tools.ToolBox( tool_configs, self.config.tool_path, self )
-        self.reindex_tool_search()
+        with self._toolbox_lock:
+            old_toolbox = self.toolbox
+            self.toolbox = tools.ToolBox( tool_configs, self.config.tool_path, self )
+            self.reindex_tool_search()
+            if old_toolbox:
+                old_toolbox.shutdown()
 
     def _configure_toolbox( self ):
         from galaxy.managers.citations import CitationsManager
@@ -762,6 +767,8 @@ class ConfiguresGalaxyMixin:
         from galaxy.tools.toolbox.cache import ToolCache
         self.tool_cache = ToolCache()
 
+        self._toolbox_lock = threading.Lock()
+        self.toolbox = None
         self.reload_toolbox()
 
         from galaxy.tools.deps import containers

lib/galaxy/tools/toolbox/base.py

@@ -28,7 +28,8 @@ from .lineages import LineageMap
 from .tags import tool_tag_manager
 
 from .filters import FilterFactory
-from .watcher import get_watcher
+from .watcher import get_tool_watcher
+from .watcher import get_tool_conf_watcher
 
 # Extra tool dependency not used by AbstractToolBox but by
 # BaseGalaxyToolBox
@@ -72,7 +73,8 @@ class AbstractToolBox( Dictifiable, ManagesIntegratedToolPanelMixin, object ):
         # (e.g., shed_tool_conf.xml) files include the tool_path attribute within the <toolbox> tag.
         self._tool_root_dir = tool_root_dir
         self.app = app
-        self._tool_watcher = get_watcher( self, app.config )
+        self._tool_watcher = get_tool_watcher( self, app.config )
+        self._tool_conf_watcher = get_tool_conf_watcher( lambda: app.reload_toolbox() )
         self._filter_factory = FilterFactory( self )
         self._tool_tag_manager = tool_tag_manager( app )
         self._init_tools_from_configs( config_filenames )
@@ -154,6 +156,9 @@ class AbstractToolBox( Dictifiable, ManagesIntegratedToolPanelMixin, object ):
                                         config_elems=config_elems )
             self._dynamic_tool_confs.append( shed_tool_conf_dict )
 
+        if tool_conf_source.parse_monitor():
+            self._tool_conf_watcher.watch_file(config_filename)
+
     def load_item( self, item, tool_path, panel_dict=None, integrated_panel_dict=None, load_panel_dict=True, guid=None, index=None, internal=False ):
         item = ensure_tool_conf_item(item)
         item_type = item.type
@@ -1003,6 +1008,21 @@ class AbstractToolBox( Dictifiable, ManagesIntegratedToolPanelMixin, object ):
 
         return rval
 
+    def shutdown(self):
+        exception = None
+        try:
+            self._tool_watcher.shutdown()
+        except Exception as e:
+            exception = e
+
+        try:
+            self._tool_conf_watcher.shutdown()
+        except Exception as e:
+            exception = exception or e
+
+        if exception:
+            raise exception
+
     def _lineage_in_panel( self, panel_dict, tool=None, tool_lineage=None ):
         """ If tool with same lineage already in panel (or section) - find
         and return it. Otherwise return None.

lib/galaxy/tools/toolbox/parser.py

@@ -1,9 +1,11 @@
 from abc import ABCMeta
 from abc import abstractmethod
 
-from galaxy.util import parse_xml
+from galaxy.util import parse_xml, string_as_bool
 import yaml
 
+DEFAULT_MONITOR = False
+
 
 class ToolConfSource(object):
     """ This interface represents an abstract source to parse tool
@@ -21,6 +23,11 @@ class ToolConfSource(object):
         """ Return tool_path for tools in this toolbox.
         """
 
+    def parse_monitor(self):
+        """ Monitor the toolbox configuration source for changes and
+        reload. """
+        return DEFAULT_MONITOR
+
 
 class XmlToolConfSource(ToolConfSource):
 
@@ -34,6 +41,9 @@ class XmlToolConfSource(ToolConfSource):
     def parse_items(self):
         return map(ensure_tool_conf_item, self.root.getchildren())
 
+    def parse_monitor(self):
+        return string_as_bool(self.root.get('monitor', DEFAULT_MONITOR))
+
 
 class YamlToolConfSource(ToolConfSource):
 
@@ -48,6 +58,9 @@ class YamlToolConfSource(ToolConfSource):
     def parse_items(self):
         return map(ToolConfItem.from_dict, self.as_dict.get('items'))
 
+    def parse_monitor(self):
+        return self.as_dict.get('monitor', DEFAULT_MONITOR)
+
 
 class ToolConfItem(object):
     """ This interface represents an abstract source to parse tool

lib/galaxy/tools/toolbox/watcher.py

@@ -1,4 +1,8 @@
+import logging
 import os.path
+import threading
+import time
+
 try:
     from watchdog.events import FileSystemEventHandler
     from watchdog.observers import Observer
@@ -9,34 +13,130 @@ except ImportError:
     PollingObserver = object
     can_watch = False
 
-import logging
 log = logging.getLogger( __name__ )
 
 
-def get_watcher(toolbox, config):
-    watch_tools_val = str( getattr(config, "watch_tools", False) ).lower()
-    if watch_tools_val in ( 'true', 'yes', 'on' ):
-        return ToolWatcher(toolbox)
-    elif watch_tools_val == "auto":
-        try:
-            return ToolWatcher(toolbox)
-        except Exception:
-            log.info("Failed to load ToolWatcher (watchdog is likely unavailable) - proceeding without tool monitoring.")
-            return NullWatcher()
-    elif watch_tools_val == "polling":
-        log.info("Using less ineffecient polling toolbox watcher.")
-        return ToolWatcher(toolbox, observer_class=PollingObserver)
+def get_observer_class(config_value, default, monitor_what_str):
+    """
+    """
+    config_value = config_value or default
+    config_value = str(config_value).lower()
+    if config_value in ("true", "yes", "on", "auto"):
+        expect_observer = config_value != "auto"
+        observer_class = Observer
+    elif config_value == "polling":
+        expect_observer = True
+        observer_class = PollingObserver
+    else:
+        expect_observer = False
+        observer_class = None
+
+    if observer_class is None:
+        message = "Watchdog library unavailble, cannot monitor %s." % monitor_what_str
+        log.info(message)
+        if expect_observer:
+            raise Exception(message)
+
+    return observer_class
+
+
+def get_tool_conf_watcher(reload_callback):
+    return ToolConfWatcher(reload_callback)
+
+
+def get_tool_watcher(toolbox, config):
+    config_value = getattr(config, "watch_tools", None)
+    observer_class = get_observer_class(config_value, default="False", monitor_what_str="tools")
+
+    if observer_class is not None:
+        return ToolWatcher(toolbox, observer_class=observer_class)
     else:
         return NullWatcher()
 
 
+class ToolConfWatcher(object):
+
+    def __init__(self, reload_callback):
+        self.paths = {}
+        self._active = False
+        self._lock = threading.Lock()
+        self.thread = threading.Thread(target=self.check)
+        self.thread.daemon = True
+        self.event_handler = ToolConfFileEventHandler(reload_callback)
+
+    def start(self):
+        if not self._active:
+            self._active = True
+            self.thread.start()
+
+    def shutdown(self):
+        self._active = False
+        self.thread.join()
+
+    def check(self):
+        while self._active:
+            do_reload = False
+            with self._lock:
+                paths = list(self.paths.keys())
+            for path in paths:
+                if not os.path.exists(path):
+                    continue
+                mod_time = self.paths[path]
+                new_mod_time = None
+                if os.path.exists(path):
+                    new_mod_time = time.ctime(os.path.getmtime(path))
+                if new_mod_time != mod_time:
+                    self.paths[path] = new_mod_time
+                    do_reload = True
+
+            if do_reload:
+                t = threading.Thread(target=lambda: self.event_handler.on_any_event(None))
+                t.daemon = True
+                t.start()
+            time.sleep(1)
+
+    def monitor(self, path):
+        mod_time = None
+        if os.path.exists(path):
+            mod_time = time.ctime(os.path.getmtime(path))
+        with self._lock:
+            self.paths[path] = mod_time
+        self.start()
+
+    def watch_file(self, tool_conf_file):
+        self.monitor(tool_conf_file)
+
+
+class NullToolConfWatcher(object):
+
+    def start(self):
+        pass
+
+    def shutdown(self):
+        pass
+
+    def monitor(self, conf_path):
+        pass
+
+    def watch_file(self, tool_file, tool_id):
+        pass
+
+
+class ToolConfFileEventHandler(FileSystemEventHandler):
+
+    def __init__(self, reload_callback):
+        self.reload_callback = reload_callback
+
+    def on_any_event(self, event):
+        self._handle(event)
+
+    def _handle(self, event):
+        self.reload_callback()
+
+
 class ToolWatcher(object):
 
-    def __init__(self, toolbox, observer_class=None):
-        if not can_watch:
-            raise Exception("Watchdog library unavailble, cannot watch tools.")
-        if observer_class is None:
-            observer_class = Observer
+    def __init__(self, toolbox, observer_class):
         self.toolbox = toolbox
         self.tool_file_ids = {}
         self.tool_dir_callbacks = {}
@@ -48,6 +148,10 @@ class ToolWatcher(object):
     def start(self):
         self.observer.start()
 
+    def shutdown(self):
+        self.observer.stop()
+        self.observer.join()
+
     def monitor(self, dir):
         self.observer.schedule(self.event_handler, dir, recursive=False)
 

test/unit/tools/test_watcher.py

@@ -17,13 +17,43 @@ def test_watcher():
         tool_path = path.join(t, "test.xml")
         toolbox = Toolbox()
         open(tool_path, "w").write("a")
-        tool_watcher = watcher.get_watcher(toolbox, bunch.Bunch(
+        tool_watcher = watcher.get_tool_watcher(toolbox, bunch.Bunch(
             watch_tools=True
         ))
         tool_watcher.watch_file(tool_path, "cool_tool")
+        assert not toolbox.was_reloaded("cool_tool")
         open(tool_path, "w").write("b")
-        time.sleep(2)
-        toolbox.assert_reloaded("cool_tool")
+        wait_for_reload(lambda: toolbox.was_reloaded("cool_tool"))
+        tool_watcher.shutdown()
+        assert not tool_watcher.observer.is_alive()
+
+
+def test_tool_conf_watcher():
+    if not watcher.can_watch:
+        from nose.plugins.skip import SkipTest
+        raise SkipTest()
+
+    callback = CallbackRecorder()
+    conf_watcher = watcher.get_tool_conf_watcher(callback.call)
+
+    with __test_directory() as t:
+        tool_conf_path = path.join(t, "test_conf.xml")
+        conf_watcher.watch_file(tool_conf_path)
+
+        open(tool_conf_path, "w").write("b")
+        wait_for_reload(lambda: callback.called)
+        conf_watcher.shutdown()
+        assert not conf_watcher.thread.is_alive()
+
+
+def wait_for_reload(check):
+    reloaded = False
+    for i in range(10):
+        reloaded = check()
+        if reloaded:
+            break
+        time.sleep(.2)
+    assert reloaded
 
 
 class Toolbox(object):
@@ -34,8 +64,8 @@ class Toolbox(object):
     def reload_tool_by_id( self, tool_id ):
         self.reloaded[ tool_id ] = True
 
-    def assert_reloaded(self, tool_id):
-        assert self.reloaded.get( tool_id, False )
+    def was_reloaded(self, tool_id):
+        return self.reloaded.get( tool_id, False )
 
 
 class CallbackRecorder(object):