diff options
Diffstat (limited to 'src/lib/Bcfg2/Server/Core.py')
| -rw-r--r-- | src/lib/Bcfg2/Server/Core.py | 220 |
1 files changed, 118 insertions, 102 deletions
diff --git a/src/lib/Bcfg2/Server/Core.py b/src/lib/Bcfg2/Server/Core.py index c69e8b055..b0cb4cc87 100644 --- a/src/lib/Bcfg2/Server/Core.py +++ b/src/lib/Bcfg2/Server/Core.py @@ -20,7 +20,9 @@ from itertools import chain from Bcfg2.Server.Cache import Cache from Bcfg2.Options import get_option_parser, SERVER_FAM_IGNORE from Bcfg2.Compat import xmlrpclib # pylint: disable=W0622 -from Bcfg2.Server.Plugin import PluginInitError, PluginExecutionError +from Bcfg2.Server.Plugin.exceptions import * # pylint: disable=W0401,W0614 +from Bcfg2.Server.Plugin.interfaces import * # pylint: disable=W0401,W0614 +from Bcfg2.Server.Plugin import track_statistics try: import psyco @@ -93,6 +95,7 @@ class BaseCore(object): .. automethod:: _block .. ----- .. automethod:: _file_monitor_thread + .. automethod:: _perflog_thread """ #: The Bcfg2 options dict self.setup = get_option_parser() @@ -179,6 +182,9 @@ class BaseCore(object): #: the first one loaded wins. self.plugin_blacklist = {} + #: The Metadata plugin + self.metadata = None + #: Revision of the Bcfg2 specification. This will be sent to #: the client in the configuration, and can be set by a #: :class:`Bcfg2.Server.Plugin.interfaces.Version` plugin. @@ -237,71 +243,6 @@ class BaseCore(object): self.logger.error("Failed to set ownership of database " "at %s: %s" % (db_settings['NAME'], err)) - if '' in self.setup['plugins']: - self.setup['plugins'].remove('') - - for plugin in self.setup['plugins']: - if not plugin in self.plugins: - self.init_plugin(plugin) - # Remove blacklisted plugins - for plugin, blacklist in list(self.plugin_blacklist.items()): - if len(blacklist) > 0: - self.logger.error("The following plugins conflict with %s;" - "Unloading %s" % (plugin, blacklist)) - for plug in blacklist: - del self.plugins[plug] - - # Log experimental plugins - expl = [plug for plug in list(self.plugins.values()) - if plug.experimental] - if expl: - self.logger.info("Loading experimental plugin(s): %s" % - (" ".join([x.name for x in expl]))) - self.logger.info("NOTE: Interfaces subject to change") - - # Log deprecated plugins - depr = [plug for plug in list(self.plugins.values()) - if plug.deprecated] - if depr: - self.logger.info("Loading deprecated plugin(s): %s" % - (" ".join([x.name for x in depr]))) - - # Find the metadata plugin and set self.metadata - mlist = self.plugins_by_type(Bcfg2.Server.Plugin.Metadata) - if len(mlist) >= 1: - #: The Metadata plugin - self.metadata = mlist[0] - if len(mlist) > 1: - self.logger.error("Multiple Metadata plugins loaded; " - "using %s" % self.metadata) - else: - self.logger.error("No Metadata plugin loaded; " - "failed to instantiate Core") - raise CoreInitError("No Metadata Plugin") - - #: The list of plugins that handle - #: :class:`Bcfg2.Server.Plugin.interfaces.Statistics` - self.statistics = self.plugins_by_type(Bcfg2.Server.Plugin.Statistics) - - #: The list of plugins that implement the - #: :class:`Bcfg2.Server.Plugin.interfaces.PullSource` - #: interface - self.pull_sources = \ - self.plugins_by_type(Bcfg2.Server.Plugin.PullSource) - - #: The list of - #: :class:`Bcfg2.Server.Plugin.interfaces.Generator` plugins - self.generators = self.plugins_by_type(Bcfg2.Server.Plugin.Generator) - - #: The list of plugins that handle - #: :class:`Bcfg2.Server.Plugin.interfaces.Structure` - #: generation - self.structures = self.plugins_by_type(Bcfg2.Server.Plugin.Structure) - - #: The list of plugins that implement the - #: :class:`Bcfg2.Server.Plugin.interfaces.Connector` interface - self.connectors = self.plugins_by_type(Bcfg2.Server.Plugin.Connector) - #: The CA that signed the server cert self.ca = self.setup['ca'] @@ -319,6 +260,12 @@ class BaseCore(object): threading.Thread(name="%sFAMThread" % self.setup['filemonitor'], target=self._file_monitor_thread) + self.perflog_thread = None + if self.setup['perflog']: + self.perflog_thread = \ + threading.Thread(name="PerformanceLoggingThread", + target=self._perflog_thread) + #: A :func:`threading.Lock` for use by #: :func:`Bcfg2.Server.FileMonitor.FileMonitor.handle_event_set` self.lock = threading.Lock() @@ -327,10 +274,6 @@ class BaseCore(object): #: metadata self.metadata_cache = Cache() - if self.debug_flag: - # enable debugging on everything else. - self.plugins[plugin].set_debug(self.debug_flag) - def plugins_by_type(self, base_cls): """ Return a list of loaded plugins that match the passed type. @@ -351,11 +294,23 @@ class BaseCore(object): if isinstance(plugin, base_cls)], key=lambda p: (p.sort_order, p.name)) + def _perflog_thread(self): + """ The thread that periodically logs performance statistics + to syslog. """ + self.logger.debug("Performance logging thread starting") + while not self.terminate.isSet(): + self.terminate.wait(self.setup['perflog_interval']) + for name, stats in self.get_statistics(None).items(): + self.logger.info("Performance statistics: " + "%s min=%.06f, max=%.06f, average=%.06f, " + "count=%d" % ((name, ) + stats)) + def _file_monitor_thread(self): """ The thread that runs the :class:`Bcfg2.Server.FileMonitor.FileMonitor`. This also queries :class:`Bcfg2.Server.Plugin.interfaces.Version` plugins for the current revision of the Bcfg2 repo. """ + self.logger.debug("File monitor thread starting") famfd = self.fam.fileno() terminate = self.terminate while not terminate.isSet(): @@ -374,7 +329,7 @@ class BaseCore(object): def _update_vcs_revision(self): """ Update the revision of the current configuration on-disk from the VCS plugin """ - for plugin in self.plugins_by_type(Bcfg2.Server.Plugin.Version): + for plugin in self.plugins_by_type(Version): try: newrev = plugin.get_revision() if newrev != self.revision: @@ -386,6 +341,59 @@ class BaseCore(object): (plugin.name, sys.exc_info()[1])) self.revision = '-1' + def load_plugins(self): + """ Load all plugins, setting + :attr:`Bcfg2.Server.Core.BaseCore.plugins` and + :attr:`Bcfg2.Server.Core.BaseCore.metadata` as side effects. + This does not start plugin threads; that is done later, in + :func:`Bcfg2.Server.Core.BaseCore.run` """ + while '' in self.setup['plugins']: + self.setup['plugins'].remove('') + + for plugin in self.setup['plugins']: + if not plugin in self.plugins: + self.init_plugin(plugin) + + # Remove blacklisted plugins + for plugin, blacklist in list(self.plugin_blacklist.items()): + if len(blacklist) > 0: + self.logger.error("The following plugins conflict with %s;" + "Unloading %s" % (plugin, blacklist)) + for plug in blacklist: + del self.plugins[plug] + + # Log deprecated and experimental plugins + expl = [] + depr = [] + for plug in list(self.plugins.values()): + if plug.experimental: + expl.append(plug) + if plug.deprecated: + depr.append(plug) + if expl: + self.logger.info("Loading experimental plugin(s): %s" % + (" ".join([x.name for x in expl]))) + self.logger.info("NOTE: Interfaces subject to change") + if depr: + self.logger.info("Loading deprecated plugin(s): %s" % + (" ".join([x.name for x in depr]))) + + # Find the metadata plugin and set self.metadata + mlist = self.plugins_by_type(Metadata) + if len(mlist) >= 1: + self.metadata = mlist[0] + if len(mlist) > 1: + self.logger.error("Multiple Metadata plugins loaded; using %s" + % self.metadata) + else: + self.logger.error("No Metadata plugin loaded; " + "failed to instantiate Core") + raise CoreInitError("No Metadata Plugin") + + if self.debug_flag: + # enable debugging on plugins + self.plugins[plugin].set_debug(self.debug_flag) + def init_plugin(self, plugin): """ Import and instantiate a single plugin. The plugin is stored to :attr:`plugins`. @@ -399,7 +407,7 @@ class BaseCore(object): self.logger.debug("Loading plugin %s" % plugin) try: mod = getattr(__import__("Bcfg2.Server.Plugins.%s" % - (plugin)).Server.Plugins, plugin) + (plugin)).Server.Plugins, plugin) except ImportError: try: mod = __import__(plugin, globals(), locals(), @@ -422,6 +430,10 @@ class BaseCore(object): except PluginInitError: self.logger.error("Failed to instantiate plugin %s" % plugin, exc_info=1) + except OSError: + err = sys.exc_info()[1] + self.logger.error("Failed to add a file monitor while " + "instantiating plugin %s: %s" % (plugin, err)) except: self.logger.error("Unexpected instantiation failure for plugin %s" % plugin, exc_info=1) @@ -468,8 +480,7 @@ class BaseCore(object): metadata.hostname)) start = time.time() try: - for plugin in \ - self.plugins_by_type(Bcfg2.Server.Plugin.ClientRunHooks): + for plugin in self.plugins_by_type(ClientRunHooks): try: getattr(plugin, hook)(metadata) except AttributeError: @@ -500,11 +511,10 @@ class BaseCore(object): :type data: list of lxml.etree._Element objects """ self.logger.debug("Validating structures for %s" % metadata.hostname) - for plugin in \ - self.plugins_by_type(Bcfg2.Server.Plugin.StructureValidator): + for plugin in self.plugins_by_type(StructureValidator): try: plugin.validate_structures(metadata, data) - except Bcfg2.Server.Plugin.ValidationError: + except ValidationError: err = sys.exc_info()[1] self.logger.error("Plugin %s structure validation failed: %s" % (plugin.name, err)) @@ -527,10 +537,10 @@ class BaseCore(object): :type data: list of lxml.etree._Element objects """ self.logger.debug("Validating goals for %s" % metadata.hostname) - for plugin in self.plugins_by_type(Bcfg2.Server.Plugin.GoalValidator): + for plugin in self.plugins_by_type(GoalValidator): try: plugin.validate_goals(metadata, data) - except Bcfg2.Server.Plugin.ValidationError: + except ValidationError: err = sys.exc_info()[1] self.logger.error("Plugin %s goal validation failed: %s" % (plugin.name, err.message)) @@ -548,8 +558,9 @@ class BaseCore(object): :returns: list of :class:`lxml.etree._Element` objects """ self.logger.debug("Getting structures for %s" % metadata.hostname) - structures = list(chain(*[struct.BuildStructures(metadata) - for struct in self.structures])) + structures = list( + chain(*[struct.BuildStructures(metadata) + for struct in self.plugins_by_type(Structure)])) sbundles = [b.get('name') for b in structures if b.tag == 'Bundle'] missing = [b for b in metadata.bundles if b not in sbundles] if missing: @@ -634,8 +645,9 @@ class BaseCore(object): self.logger.error("Falling back to %s:%s" % (entry.tag, entry.get('name'))) - glist = [gen for gen in self.generators if - entry.get('name') in gen.Entries.get(entry.tag, {})] + generators = self.plugins_by_type(Generator) + glist = [gen for gen in generators + if entry.get('name') in gen.Entries.get(entry.tag, {})] if len(glist) == 1: return glist[0].Entries[entry.tag][entry.get('name')](entry, metadata) @@ -643,8 +655,8 @@ class BaseCore(object): generators = ", ".join([gen.name for gen in glist]) self.logger.error("%s %s served by multiple generators: %s" % (entry.tag, entry.get('name'), generators)) - g2list = [gen for gen in self.generators if - gen.HandlesEntry(entry, metadata)] + g2list = [gen for gen in generators + if gen.HandlesEntry(entry, metadata)] try: if len(g2list) == 1: return g2list[0].HandleEntry(entry, metadata) @@ -671,7 +683,7 @@ class BaseCore(object): revision=self.revision) try: meta = self.build_metadata(client) - except Bcfg2.Server.Plugin.MetadataConsistencyError: + except MetadataConsistencyError: self.logger.error("Metadata consistency error for client %s" % client) return lxml.etree.Element("error", type='metadata error') @@ -718,7 +730,8 @@ class BaseCore(object): :type event: Bcfg2.Server.FileMonitor.Event """ if event.filename != self.cfile: - print("Got event for unknown file: %s" % event.filename) + self.logger.error("Got event for unknown file: %s" % + event.filename) return if event.code2str() == 'deleted': return @@ -755,11 +768,15 @@ class BaseCore(object): return False try: + self.load_plugins() + self.fam.start() self.fam_thread.start() self.fam.AddMonitor(self.cfile, self) + if self.perflog_thread is not None: + self.perflog_thread.start() - for plug in self.plugins_by_type(Bcfg2.Server.Plugin.Threaded): + for plug in self.plugins_by_type(Threaded): plug.start_threads() except: self.shutdown() @@ -801,7 +818,7 @@ class BaseCore(object): """ self.logger.debug("Getting decision list for %s" % metadata.hostname) result = [] - for plugin in self.plugins_by_type(Bcfg2.Server.Plugin.Decision): + for plugin in self.plugins_by_type(Decision): try: result.extend(plugin.GetDecisions(metadata, mode)) except: @@ -862,7 +879,7 @@ class BaseCore(object): """ if not hasattr(self, 'metadata'): # some threads start before metadata is even loaded - raise Bcfg2.Server.Plugin.MetadataRuntimeError + raise MetadataRuntimeError("Metadata not loaded yet") if self.metadata_cache_mode == 'initial': # the Metadata plugin handles loading the cached data if # we're only caching the initial metadata object @@ -872,10 +889,11 @@ class BaseCore(object): if not imd: self.logger.debug("Building metadata for %s" % client_name) imd = self.metadata.get_initial_metadata(client_name) - for conn in self.connectors: + connectors = self.plugins_by_type(Connector) + for conn in connectors: grps = conn.get_additional_groups(imd) self.metadata.merge_additional_groups(imd, grps) - for conn in self.connectors: + for conn in connectors: data = conn.get_additional_data(imd) self.metadata.merge_additional_data(imd, conn.name, data) imd.query.by_name = self.build_metadata @@ -896,7 +914,7 @@ class BaseCore(object): meta = self.build_metadata(client_name) state = statistics.find(".//Statistics") if state.get('version') >= '2.0': - for plugin in self.statistics: + for plugin in self.plugins_by_type(Statistics): try: plugin.process_statistics(meta, statistics) except: @@ -938,11 +956,11 @@ class BaseCore(object): meta = self.build_metadata(client) else: meta = None - except Bcfg2.Server.Plugin.MetadataConsistencyError: + except MetadataConsistencyError: err = sys.exc_info()[1] self.critical_error("Client metadata resolution error for %s: %s" % (address[0], err)) - except Bcfg2.Server.Plugin.MetadataRuntimeError: + except MetadataRuntimeError: err = sys.exc_info()[1] self.critical_error('Metadata system runtime failure for %s: %s' % (address[0], err)) @@ -1038,8 +1056,7 @@ class BaseCore(object): version)) try: self.metadata.set_version(client, version) - except (Bcfg2.Server.Plugin.MetadataConsistencyError, - Bcfg2.Server.Plugin.MetadataRuntimeError): + except (MetadataConsistencyError, MetadataRuntimeError): err = sys.exc_info()[1] self.critical_error("Unable to set version for %s: %s" % (client, err)) @@ -1059,7 +1076,7 @@ class BaseCore(object): client, metadata = self.resolve_client(address, cleanup_cache=True) self.logger.debug("Getting probes for %s" % client) try: - for plugin in self.plugins_by_type(Bcfg2.Server.Plugin.Probing): + for plugin in self.plugins_by_type(Probing): for probe in plugin.GetProbes(metadata): resp.append(probe) return lxml.etree.tostring(resp, @@ -1129,11 +1146,10 @@ class BaseCore(object): self.logger.debug("%s sets its profile to %s" % (client, profile)) try: self.metadata.set_profile(client, profile, address) - except (Bcfg2.Server.Plugin.MetadataConsistencyError, - Bcfg2.Server.Plugin.MetadataRuntimeError): + except (MetadataConsistencyError, MetadataRuntimeError): err = sys.exc_info()[1] self.critical_error("Unable to assert profile for %s: %s" % - (client, err)) + (client, err)) return True @exposed @@ -1152,7 +1168,7 @@ class BaseCore(object): config = self.BuildConfiguration(client) return lxml.etree.tostring(config, xml_declaration=False).decode('UTF-8') - except Bcfg2.Server.Plugin.MetadataConsistencyError: + except MetadataConsistencyError: self.critical_error("Metadata consistency failure for %s" % client) @exposed |