From e226690c555e87ac13f5d2543d197864e4d0270c Mon Sep 17 00:00:00 2001 From: Narayan Desai Date: Fri, 15 May 2009 03:57:33 +0000 Subject: Merge xmlrpc interfaces with Bcfg2.Server.Core git-svn-id: https://svn.mcs.anl.gov/repos/bcfg/trunk/bcfg2@5238 ce84e21b-d406-0410-9b95-82705330c041 --- src/lib/Server/Core.py | 170 +++++++++++++++++++++++++++++++++++++++++++------ 1 file changed, 151 insertions(+), 19 deletions(-) (limited to 'src/lib/Server/Core.py') diff --git a/src/lib/Server/Core.py b/src/lib/Server/Core.py index 636542bf2..f21250d81 100644 --- a/src/lib/Server/Core.py +++ b/src/lib/Server/Core.py @@ -1,15 +1,26 @@ '''Bcfg2.Server.Core provides the runtime support for bcfg2 modules''' __revision__ = '$Revision$' -from time import time +import copy +import hashlib +import logging +import lxml.etree +import select +import threading +import time +import xmlrpclib +from Bcfg2.Component import Component, exposed from Bcfg2.Server.Plugin import PluginInitError, PluginExecutionError import Bcfg2.Server.FileMonitor - -import copy, logging, lxml.etree import Bcfg2.Server.Plugins.Metadata -logger = logging.getLogger('Bcfg2.Core') +logger = logging.getLogger('Bcfg2.Server.Core') + +def critical_error(operation): + '''Log and err, traceback and return an xmlrpc fault to client''' + logger.error(operation, exc_info=1) + raise xmlrpclib.Fault(7, "Critical unexpected failure: %s" % (operation)) try: import psyco @@ -21,12 +32,14 @@ class CoreInitError(Exception): '''This error is raised when the core cannot be initialized''' pass -class Core(object): +class Core(Component): '''The Core object is the container for all Bcfg2 Server logic, and modules''' + name = 'bcfg2' + implementation = 'bcfg2' - def __init__(self, repo, plugins, password, encoding, - filemonitor='default'): - object.__init__(self) + def __init__(self, repo, plugins, password, encoding, ca=None, + filemonitor='default', start_fam_thread=False): + Component.__init__(self) self.datastore = repo if filemonitor not in Bcfg2.Server.FileMonitor.available: logger.error("File monitor driver %s not available; forcing to default" % filemonitor) @@ -78,7 +91,24 @@ class Core(object): isinstance(plugin, Bcfg2.Server.Plugin.Structure)] self.connectors = [plugin for plugin in self.plugins.values() if \ isinstance(plugin, Bcfg2.Server.Plugin.Connector)] - + self.ca = ca + self.fam_thread = threading.Thread(target=self._file_monitor_thread) + if start_fam_thread: + self.fam_thread.start() + + def _file_monitor_thread(self): + famfd = self.fam.fileno() + while True: + try: + if famfd: + select.select([famfd], [], []) + else: + while not self.fam.pending(): + time.sleep(15) + self.fam.handle_event_set(self.lock) + except: + continue + def init_plugins(self, plugin): try: mod = getattr(__import__("Bcfg2.Server.Plugins.%s" % @@ -181,7 +211,7 @@ class Core(object): def BuildConfiguration(self, client): '''Build Configuration for client''' - start = time() + start = time.time() config = lxml.etree.Element("Configuration", version='2.0', \ revision=self.revision) try: @@ -219,17 +249,9 @@ class Core(object): logger.error("error in BindStructure", exc_info=1) self.validate_data(meta, config, Bcfg2.Server.Plugin.GoalValidator) logger.info("Generated config for %s in %.03fs" % \ - (client, time() - start)) + (client, time.time() - start)) return config - def Service(self): - '''Perform periodic update tasks''' - count = self.fam.Service() - if count: - for plugin in self.plugins.values(): - if isinstance(plugin, Bcfg2.Server.Plugin.Version): - self.revision = plugin.get_revision() - def GetDecisions(self, metadata, mode): result = [] for plugin in self.plugins.values(): @@ -267,3 +289,113 @@ class Core(object): logger.info("Client %s reported state %s" % (client_name, state.get('state'))) + # XMLRPC handlers start here + + @exposed + def GetProbes(self, address): + '''Fetch probes for a particular client''' + resp = lxml.etree.Element('probes') + try: + name = self.metadata.resolve_client(address) + meta = self.build_metadata(name) + + for plugin in [p for p in list(self.plugins.values()) \ + if isinstance(p, Bcfg2.Server.Plugin.Probing)]: + for probe in plugin.GetProbes(meta): + resp.append(probe) + return lxml.etree.tostring(resp, encoding='UTF-8', + xml_declaration=True) + except Bcfg2.Server.Plugins.Metadata.MetadataConsistencyError: + warning = 'Client metadata resolution error for %s; check server log' % address[0] + self.logger.warning(warning) + raise xmlrpclib.Fault(6, warning) + except: + critical_error("error determining client probes") + + @exposed + def RecvProbeData(self, address, probedata): + '''Receive probe data from clients''' + try: + name = self.metadata.resolve_client(address) + meta = self.build_metadata(name) + except Bcfg2.Server.Plugins.Metadata.MetadataConsistencyError: + warning = 'metadata consistency error' + self.logger.warning(warning) + raise xmlrpclib.Fault(6, warning) + # clear dynamic groups + self.metadata.cgroups[meta.hostname] = [] + try: + xpdata = lxml.etree.XML(probedata) + except: + self.logger.error("Failed to parse probe data from client %s" % \ + (address[0])) + return False + + sources = [] + [sources.append(data.get('source')) for data in xpdata + if data.get('source') not in sources] + for source in sources: + if source not in self.plugins: + self.logger.warning("Failed to locate plugin %s" % (source)) + continue + dl = [data for data in xpdata if data.get('source') == source] + try: + self.plugins[source].ReceiveData(meta, dl) + except: + logger.error("Failed to process probe data from client %s" % \ + (address[0]), exc_info=1) + return True + + @exposed + def AssertProfile(self, address, profile): + '''Set profile for a client''' + try: + client = self.metadata.resolve_client(address) + self.metadata.set_profile(client, profile, address) + except (Bcfg2.Server.Plugins.Metadata.MetadataConsistencyError, + Bcfg2.Server.Plugins.Metadata.MetadataRuntimeError): + warning = 'metadata consistency error' + self.logger.warning(warning) + raise xmlrpclib.Fault(6, warning) + return True + + @exposed + def GetConfig(self, address, checksum=False): + '''Build config for a client''' + try: + client = self.metadata.resolve_client(address) + config = self.BuildConfiguration(client) + if checksum: + for cfile in config.findall('.//ConfigFile'): + if cfile.text != None: + csum = hashlib.md5() + csum.update(cfile.text) + cfile.set('checksum', csum.hexdigest()) + cfile.text = None + return lxml.etree.tostring(config, encoding='UTF-8', + xml_declaration=True) + except Bcfg2.Server.Plugins.Metadata.MetadataConsistencyError: + self.logger.warning("Metadata consistency failure for %s" % (address)) + raise xmlrpclib.Fault(6, "Metadata consistency failure") + + @exposed + def RecvStats(self, address, stats): + '''Act on statistics upload''' + sdata = lxml.etree.XML(stats) + client = self.metadata.resolve_client(address) + self.process_statistics(client, sdata) + return "" + + def authenticate(self, cert, user, password, address): + if self.ca: + acert = cert + else: + # no ca, so no cert validation can be done + acert = None + return self.metadata.AuthenticateConnection(acert, user, password, address) + + @exposed + def GetDecisionList(self, address, mode): + client = self.metadata.resolve_client(address) + meta = self.build_metadata(client) + return self.GetDecisions(meta, mode) -- cgit v1.2.3-1-g7c22