Merge pull request #3821 from mvdbeek/repository_updates

Merge ToolConfWatchers and separate from Toolbox
This commit is contained in:
John Chilton
2017-04-07 12:34:22 -04:00
committed by GitHub
10 changed files with 146 additions and 148 deletions
+1
View File
@@ -184,6 +184,7 @@ lib/galaxy/webapps/galaxy/controllers/requests.py
lib/galaxy/webapps/galaxy/controllers/search.py
lib/galaxy/webapps/galaxy/controllers/tool_runner.py
lib/galaxy/webapps/galaxy/controllers/userskeys.py
lib/galaxy/webapps/galaxy/config_watchers.py
lib/galaxy/webapps/galaxy/__init__.py
lib/galaxy/webapps/__init__.py
lib/galaxy/webapps/reports/config.py
+6
View File
@@ -17,10 +17,12 @@ from galaxy.visualization.data_providers.registry import DataProviderRegistry
from galaxy.visualization.plugins.registry import VisualizationsRegistry
from galaxy.tools.special_tools import load_lib_tools
from galaxy.tours import ToursRegistry
from galaxy.webapps.galaxy.config_watchers import ConfigWatchers
from galaxy.webhooks import WebhooksRegistry
from galaxy.sample_tracking import external_service_types
from galaxy.openid.providers import OpenIDProviders
from galaxy.tools.data_manager.manager import DataManagers
from galaxy.tools.toolbox.cache import ToolCache
from galaxy.jobs import metrics as job_metrics
from galaxy.web.proxy import ProxyManager
from galaxy.web.stack import application_stack_instance
@@ -96,6 +98,10 @@ class UniverseApplication( object, config.ConfiguresGalaxyMixin ):
# Initialize the job management configuration
self.job_config = jobs.JobConfiguration(self)
# Setup a Tool Cache
self.tool_cache = ToolCache()
# Watch various config files for immediate reload
self.watchers = ConfigWatchers(self)
self._configure_toolbox()
# Load Data Manager
-2
View File
@@ -910,11 +910,9 @@ class ConfiguresGalaxyMixin:
from galaxy import tools
from galaxy.managers.citations import CitationsManager
from galaxy.tools.deps import containers
from galaxy.tools.toolbox.cache import ToolCache
from galaxy.tools.toolbox.lineages.tool_shed import ToolVersionCache
self.citations_manager = CitationsManager( self )
self.tool_cache = ToolCache()
self.tool_version_cache = ToolVersionCache(self)
self._toolbox_lock = threading.RLock()
+2 -2
View File
@@ -106,7 +106,7 @@ def _get_new_toolbox(app):
if app.config.migrated_tools_config not in tool_configs:
tool_configs.append(app.config.migrated_tools_config)
start = time.time()
new_toolbox = tools.ToolBox(tool_configs, app.config.tool_path, app, app.toolbox._tool_conf_watcher)
new_toolbox = tools.ToolBox(tool_configs, app.config.tool_path, app)
new_toolbox.data_manager_tools = app.toolbox.data_manager_tools
app.datatypes_registry.load_datatype_converters(new_toolbox, use_cached=True)
load_lib_tools(new_toolbox)
@@ -124,7 +124,7 @@ def reload_data_managers(app, **kwargs):
app._configure_tool_data_tables(from_shed_config=False)
reload_tool_data_tables(app)
reload_count = app.data_managers._reload_count
app.data_managers = DataManagers(app, conf_watchers=app.data_managers.conf_watchers)
app.data_managers = DataManagers(app)
app.data_managers._reload_count = reload_count + 1
+1 -8
View File
@@ -29,10 +29,7 @@ from galaxy import (
)
from galaxy.datatypes.metadata import JobExternalOutputMetadataWrapper
from galaxy.managers import histories
from galaxy.queue_worker import (
reload_toolbox,
send_control_task
)
from galaxy.queue_worker import send_control_task
from galaxy.tools.actions import DefaultToolAction
from galaxy.tools.actions.data_manager import DataManagerToolAction
from galaxy.tools.actions.data_source import DataSourceToolAction
@@ -195,12 +192,8 @@ class ToolBox( BaseGalaxyToolBox ):
config_filenames=config_filenames,
tool_root_dir=tool_root_dir,
app=app,
tool_conf_watcher=tool_conf_watcher
)
def handle_reload_toolbox(self):
reload_toolbox(self.app)
def handle_panel_update(self, section_dict):
"""
Sends a panel update to all threads/processes.
+48 -65
View File
@@ -7,14 +7,9 @@ from six import string_types
from galaxy import util
from galaxy.queue_worker import (
reload_data_managers,
send_control_task
)
from galaxy.tools.data import TabularToolDataTable
from galaxy.tools.toolbox.watcher import (
get_tool_conf_watcher,
get_tool_data_dir_watcher
)
from galaxy.util.odict import odict
from galaxy.util.template import fill_template
from tool_shed.util import (
@@ -30,7 +25,7 @@ DEFAULT_VALUE_TRANSLATION_TYPE = 'template'
class DataManagers( object ):
def __init__( self, app, xml_filename=None, conf_watchers=None ):
def __init__( self, app, xml_filename=None):
self.app = app
self.data_managers = odict()
self.managed_data_tables = odict()
@@ -43,31 +38,6 @@ class DataManagers( object ):
self.load_from_xml( filename )
if self.app.config.shed_data_manager_config_file:
self.load_from_xml( self.app.config.shed_data_manager_config_file, store_tool_path=False, replace_existing=True )
if conf_watchers:
self.conf_watchers = conf_watchers
else:
self.conf_watchers = self.get_conf_watchers()
def get_conf_watchers(self):
"""
Sets up monitoring of data manager related config files.
These are the data_manager_conf.xml and shed_data_manager_conf.xml files
as well as any loc file in the tool-data directory.
"""
conf_watchers = []
conf_watchers.extend([(get_tool_conf_watcher(lambda: reload_data_managers(self.app)), filename) for filename in util.listify(self.filename) if filename])
if self.app.config.shed_data_manager_config_file:
conf_watchers.append((get_tool_conf_watcher(lambda: reload_data_managers(self.app)), self.app.config.shed_data_manager_config_file))
[watcher.watch_file(filename) for watcher, filename in conf_watchers]
tool_data_watcher = get_tool_data_dir_watcher(self.app.tool_data_tables, self.app.config)
tool_data_watcher.watch_directory(self.app.config.tool_data_path)
if self.app.config.shed_tool_data_path:
tool_data_watcher.watch_directory(self.app.config.shed_tool_data_path)
conf_watchers.append((tool_data_watcher, self.app.tool_data_tables.tool_data_path))
return [watcher for watcher, filename in conf_watchers]
def shutdown(self):
[watcher.shutdown() for watcher in self.conf_watchers]
def load_from_xml( self, xml_filename, store_tool_path=True, replace_existing=False ):
try:
@@ -166,46 +136,58 @@ class DataManager( object ):
self.version = elem.get( 'version', self.version )
tool_shed_repository_id = None
tool_guid = None
if path is None:
tool_elem = elem.find( 'tool' )
assert tool_elem is not None, "Error loading tool for data manager. Make sure that a tool_file attribute or a tool tag set has been defined:\n%s" % ( util.xml_to_string( elem ) )
path = tool_elem.get( "file", None )
tool_guid = tool_elem.get( "guid", None )
# need to determine repository info so that dependencies will work correctly
tool_shed_url = tool_elem.find( 'tool_shed' ).text
# Handle protocol changes.
tool_shed_url = common_util.get_tool_shed_url_from_tool_shed_registry( self.data_managers.app, tool_shed_url )
# The protocol is not stored in the database.
tool_shed = common_util.remove_protocol_from_tool_shed_url( tool_shed_url )
repository_name = tool_elem.find( 'repository_name' ).text
repository_owner = tool_elem.find( 'repository_owner' ).text
installed_changeset_revision = tool_elem.find( 'installed_changeset_revision' ).text
self.tool_shed_repository_info_dict = dict( tool_shed=tool_shed,
name=repository_name,
owner=repository_owner,
installed_changeset_revision=installed_changeset_revision )
tool_shed_repository = \
repository_util.get_installed_repository( self.data_managers.app,
tool_shed=tool_shed,
name=repository_name,
owner=repository_owner,
installed_changeset_revision=installed_changeset_revision )
if tool_shed_repository is None:
log.warning( 'Could not determine tool shed repository from database. This should only ever happen when running tests.' )
# we'll set tool_path manually here from shed_conf_file
tool_shed_repository_id = None
try:
tool_path = util.parse_xml( elem.get( 'shed_conf_file' ) ).getroot().get( 'tool_path', tool_path )
except Exception as e:
log.error( 'Error determining tool_path for Data Manager during testing: %s', e )
if tool_guid in self.data_managers.app.tool_cache._tool_paths_by_id:
path = self.data_managers.app.tool_cache._tool_paths_by_id[tool_guid]
tool = self.data_managers.app.tool_cache.get_tool(path)
tool_shed_repository = tool.tool_shed_repository
self.tool_shed_repository_info_dict = dict(tool_shed=tool_shed_repository.tool_shed,
name=tool_shed_repository.name,
owner=tool_shed_repository.owner,
installed_changeset_revision=tool_shed_repository.installed_changeset_revision)
tool_shed_repository_id = self.data_managers.app.security.encode_id(tool_shed_repository.id)
tool_path = ""
else:
tool_shed_repository_id = self.data_managers.app.security.encode_id( tool_shed_repository.id )
# use shed_conf_file to determine tool_path
shed_conf_file = elem.get( "shed_conf_file", None )
if shed_conf_file:
shed_conf = self.data_managers.app.toolbox.get_shed_config_dict_by_filename( shed_conf_file, None )
if shed_conf:
tool_path = shed_conf.get( "tool_path", tool_path )
tool_shed_url = tool_elem.find( 'tool_shed' ).text
# Handle protocol changes.
tool_shed_url = common_util.get_tool_shed_url_from_tool_shed_registry( self.data_managers.app, tool_shed_url )
# The protocol is not stored in the database.
tool_shed = common_util.remove_protocol_from_tool_shed_url( tool_shed_url )
repository_name = tool_elem.find( 'repository_name' ).text
repository_owner = tool_elem.find( 'repository_owner' ).text
installed_changeset_revision = tool_elem.find( 'installed_changeset_revision' ).text
self.tool_shed_repository_info_dict = dict( tool_shed=tool_shed,
name=repository_name,
owner=repository_owner,
installed_changeset_revision=installed_changeset_revision )
tool_shed_repository = \
repository_util.get_installed_repository( self.data_managers.app,
tool_shed=tool_shed,
name=repository_name,
owner=repository_owner,
installed_changeset_revision=installed_changeset_revision )
if tool_shed_repository is None:
log.warning( 'Could not determine tool shed repository from database. This should only ever happen when running tests.' )
# we'll set tool_path manually here from shed_conf_file
tool_shed_repository_id = None
try:
tool_path = util.parse_xml( elem.get( 'shed_conf_file' ) ).getroot().get( 'tool_path', tool_path )
except Exception as e:
log.error( 'Error determining tool_path for Data Manager during testing: %s', e )
else:
tool_shed_repository_id = self.data_managers.app.security.encode_id( tool_shed_repository.id )
# use shed_conf_file to determine tool_path
shed_conf_file = elem.get( "shed_conf_file", None )
if shed_conf_file:
shed_conf = self.data_managers.app.toolbox.get_shed_config_dict_by_filename( shed_conf_file, None )
if shed_conf:
tool_path = shed_conf.get( "tool_path", tool_path )
assert path is not None, "A tool file path could not be determined:\n%s" % ( util.xml_to_string( elem ) )
self.load_tool( os.path.join( tool_path, path ),
guid=tool_guid,
@@ -287,7 +269,8 @@ class DataManager( object ):
tool = toolbox.load_hidden_tool( tool_filename,
guid=guid,
data_manager_id=data_manager_id,
repository_id=tool_shed_repository_id )
repository_id=tool_shed_repository_id,
use_cached=True )
self.data_managers.app.toolbox.data_manager_tools[ tool.id ] = tool
self.tool = tool
return tool
+16 -49
View File
@@ -33,10 +33,6 @@ from .panel import (
)
from .parser import ensure_tool_conf_item, get_toolbox_parser
from .tags import tool_tag_manager
from .watcher import (
get_tool_conf_watcher,
get_tool_watcher
)
log = logging.getLogger( __name__ )
@@ -47,11 +43,10 @@ class AbstractToolBox( Dictifiable, ManagesIntegratedToolPanelMixin, object ):
workflows optionally in labelled sections.
"""
def __init__( self, config_filenames, tool_root_dir, app, tool_conf_watcher=None ):
def __init__( self, config_filenames, tool_root_dir, app ):
"""
Create a toolbox from the config files named by `config_filenames`, using
`tool_root_dir` as the base directory for finding individual tool config files.
When reloading the toolbox, tool_conf_watcher will be provided.
"""
# The _dynamic_tool_confs list contains dictionaries storing
# information about the tools defined in each shed-related
@@ -75,17 +70,11 @@ 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_tool_watcher( self, app.config )
if tool_conf_watcher:
self._tool_conf_watcher = tool_conf_watcher # Avoids (re-)starting threads in uwsgi
if hasattr(self.app, 'watchers'):
self._tool_watcher = self.app.watchers.tool_watcher
else:
if hasattr(self.app, 'tool_cache'):
# Normal galaxy instances should have a tool_cache,
# but the toolshed does not.
tool_cache = self.app.tool_cache
else:
tool_cache = None
self._tool_conf_watcher = get_tool_conf_watcher(reload_callback=lambda: self.handle_reload_toolbox(), tool_cache=tool_cache)
# Toolbox is loaded but not used during toolshed tests
self._tool_watcher = None
self._filter_factory = FilterFactory( self )
self._tool_tag_manager = tool_tag_manager( app )
self._init_tools_from_configs( config_filenames )
@@ -94,13 +83,6 @@ class AbstractToolBox( Dictifiable, ManagesIntegratedToolPanelMixin, object ):
self._load_tool_panel()
self._save_integrated_tool_panel()
def handle_reload_toolbox(self):
"""Extension-point for Galaxy-app specific reload logic.
This abstract representation of the toolbox shouldn't have details about
interacting with the rest of the Galaxy app or message queues, etc....
"""
def handle_panel_update(self, section_dict):
"""Extension-point for Galaxy-app specific reload logic.
@@ -188,10 +170,6 @@ class AbstractToolBox( Dictifiable, ManagesIntegratedToolPanelMixin, object ):
tool_path=tool_path,
config_elems=config_elems )
self._dynamic_tool_confs.append( shed_tool_conf_dict )
# This explicitly monitors shed_tool_confs, otherwise need to add <toolbox monitor="true">>
self._tool_conf_watcher.watch_file(config_filename)
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 ):
with self.app._toolbox_lock:
@@ -537,10 +515,14 @@ class AbstractToolBox( Dictifiable, ManagesIntegratedToolPanelMixin, object ):
path_template = item.get( "file" )
template_kwds = self._path_template_kwds()
path = string.Template(path_template).safe_substitute(**template_kwds)
concrete_path = os.path.join(tool_path, path)
if not os.path.exists(concrete_path):
# This is a lot faster than attempting to load a non-existing tool
raise IOError
tool_shed_repository = None
can_load_into_panel_dict = True
tool = self.load_tool_from_cache(os.path.join(tool_path, path))
tool = self.load_tool_from_cache(concrete_path)
from_cache = tool
if from_cache:
if guid and tool.id != guid:
@@ -555,9 +537,9 @@ class AbstractToolBox( Dictifiable, ManagesIntegratedToolPanelMixin, object ):
# Only load tools if the repository is not deactivated or uninstalled.
can_load_into_panel_dict = not tool_shed_repository.deleted
repository_id = self.app.security.encode_id(tool_shed_repository.id)
tool = self.load_tool(os.path.join( tool_path, path ), guid=guid, repository_id=repository_id, use_cached=False)
tool = self.load_tool(concrete_path, guid=guid, repository_id=repository_id, use_cached=False)
if not tool: # tool was not in cache and is not a tool shed tool.
tool = self.load_tool(os.path.join(tool_path, path), use_cached=False)
tool = self.load_tool(concrete_path, use_cached=False)
if string_as_bool(item.get( 'hidden', False )):
tool.hidden = True
key = 'tool_%s' % str(tool.id)
@@ -729,7 +711,7 @@ class AbstractToolBox( Dictifiable, ManagesIntegratedToolPanelMixin, object ):
elif self._looks_like_a_tool(child_path):
quick_load( child_path, async=False )
tool_loaded = True
if tool_loaded or force_watch:
if (tool_loaded or force_watch) and self._tool_watcher:
self._tool_watcher.watch_directory( directory, quick_load )
def load_tool( self, config_file, guid=None, repository_id=None, use_cached=False, **kwds ):
@@ -742,7 +724,7 @@ class AbstractToolBox( Dictifiable, ManagesIntegratedToolPanelMixin, object ):
tool = self.create_tool( config_file=config_file, repository_id=repository_id, guid=guid, **kwds )
if tool.tool_shed_repository or not guid:
self.add_tool_to_cache(tool, config_file)
if not tool.id.startswith("__"):
if not tool.id.startswith("__") and self._tool_watcher:
# do not monitor special tools written to tmp directory - no reason
# to monitor such a large directory.
self._tool_watcher.watch_file( config_file, tool.id )
@@ -941,21 +923,6 @@ 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.
@@ -1066,8 +1033,8 @@ class BaseGalaxyToolBox(AbstractToolBox):
shouldn't really depend on.
"""
def __init__(self, config_filenames, tool_root_dir, app, tool_conf_watcher=None):
super(BaseGalaxyToolBox, self).__init__(config_filenames, tool_root_dir, app, tool_conf_watcher=tool_conf_watcher)
def __init__(self, config_filenames, tool_root_dir, app):
super(BaseGalaxyToolBox, self).__init__(config_filenames, tool_root_dir, app)
self._init_dependency_manager()
@property
+9 -22
View File
@@ -80,7 +80,7 @@ class ToolConfWatcher(object):
self._lock = threading.Lock()
self.thread = threading.Thread(target=self.check, name="ToolConfWatcher.thread")
self.thread.daemon = True
self.event_handler = ToolConfFileEventHandler(reload_callback)
self.reload_callback = reload_callback
def start(self):
if not self._active:
@@ -107,8 +107,8 @@ class ToolConfWatcher(object):
hashes[path] = md5_hash_file(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:
new_mod_time = os.path.getmtime(path)
if new_mod_time > mod_time:
new_hash = md5_hash_file(path)
if hashes[path] != new_hash:
self.paths[path] = new_mod_time
@@ -120,23 +120,22 @@ class ToolConfWatcher(object):
if removed_ids:
do_reload = True
if do_reload:
with self._lock:
t = threading.Thread(target=self.event_handler.on_any_event)
t.daemon = True
t.start()
self.reload_callback()
time.sleep(1)
def monitor(self, path):
mod_time = None
if os.path.exists(path):
mod_time = time.ctime(os.path.getmtime(path))
mod_time = os.path.getmtime(path)
with self._lock:
self.paths[path] = mod_time
self.start()
if not self._active:
self.start()
def watch_file(self, tool_conf_file):
self.monitor(tool_conf_file)
self.start()
if not self._active:
self.start()
class NullToolConfWatcher(object):
@@ -154,18 +153,6 @@ class NullToolConfWatcher(object):
pass
class ToolConfFileEventHandler(FileSystemEventHandler):
def __init__(self, reload_callback):
self.reload_callback = reload_callback
def on_any_event(self, event=None):
self._handle(event)
def _handle(self, event):
self.reload_callback()
class ToolWatcher(object):
def __init__(self, toolbox, observer_class):
@@ -0,0 +1,61 @@
from galaxy.queue_worker import (
reload_data_managers,
reload_toolbox,
)
from galaxy.tools.toolbox.watcher import (
get_tool_conf_watcher,
get_tool_data_dir_watcher,
get_tool_watcher,
)
class ConfigWatchers(object):
"""Contains ToolConfWatcher, ToolWatcher and ToolDataWatcher objects."""
def __init__(self, app):
self.app = app
# ToolConfWatcher objects will watch the tool_cache if the tool_cache is passed into get_tool_conf_watcher.
# Watching the tool_cache means removing outdated items from the tool_cache.
# Only the reload_toolbox callback will re-populate the cache, so we pass the tool_cache only to the ToolConfWatcher that
# watches regular tools.
# If there are multiple ToolConfWatcher objects for the same handler or web process a race condition occurs between the two cache_cleanup functions.
# If the reload_data_managers callback wins, the cache will miss the tools that had been removed from the cache
# and will be blind to further changes in these tools.
self.tool_config_watcher = get_tool_conf_watcher(reload_callback=lambda: reload_toolbox(self.app), tool_cache=self.app.tool_cache)
self.data_manager_config_watcher = get_tool_conf_watcher(reload_callback=lambda: reload_data_managers(self.app))
self.tool_data_watcher = get_tool_data_dir_watcher(self.app.tool_data_tables, config=self.app.config)
self.tool_watcher = get_tool_watcher( self, app.config )
self.start()
def start(self):
[self.tool_config_watcher.watch_file(config) for config in self.tool_config_paths]
[self.data_manager_config_watcher.watch_file(config) for config in self.data_manager_configs]
[self.tool_data_watcher.watch_directory(tool_data_path) for tool_data_path in self.tool_data_paths]
@property
def data_manager_configs(self):
data_manager_configs = []
if hasattr(self.app.config, 'data_manager_config_file'):
data_manager_configs.append(self.app.config.data_manager_config_file)
if hasattr(self.app.config, 'shed_data_manager_config_file'):
data_manager_configs.append(self.app.config.shed_data_manager_config_file)
return data_manager_configs
@property
def tool_data_paths(self):
tool_data_paths = []
if hasattr(self.app.config, 'tool_data_path'):
tool_data_paths.append(self.app.config.tool_data_path)
if hasattr(self.app.config, 'shed_tool_data_path'):
tool_data_paths.append(self.app.config.shed_tool_data_path)
return tool_data_paths
@property
def tool_config_paths(self):
tool_config_paths = []
if hasattr(self.app.config, 'tool_configs'):
tool_config_paths = self.app.config.tool_configs
if hasattr(self.app.config, 'migrated_tools_config'):
if self.app.config.migrated_tools_config not in tool_config_paths:
tool_config_paths.append(self.app.config.migrated_tools_config)
return tool_config_paths
+2
View File
@@ -13,6 +13,7 @@ from galaxy.model.tool_shed_install import mapping
from galaxy.tools import ToolBox
from galaxy.tools.toolbox.lineages.tool_shed import ToolVersionCache
from galaxy.tools.toolbox.watcher import get_tool_conf_watcher
from galaxy.webapps.galaxy.config_watchers import ConfigWatchers
from .test_toolbox_filters import mock_trans
@@ -62,6 +63,7 @@ class BaseToolBoxTestCase( unittest.TestCase, tools_support.UsesApp, tools_supp
self.app.reindex_tool_search = self.__reindex
itp_config = os.path.join(self.test_directory, "integrated_tool_panel.xml")
self.app.config.integrated_tool_panel_config = itp_config
self.app.watchers = ConfigWatchers(self.app)
self.__toolbox = None
self.config_files = []