Multiprocessing: proxy RecvProbeData calls

This proxies RecvProbeData calls to child cores to expire the probe
cache.  The probe data itself is not relayed, just the fact that there
was probe data received from a given client.

Fixes #129.

2 files changed, 29 insertions, 5 deletions

diff --git a/src/lib/Bcfg2/Server/MultiprocessingCore.py b/src/lib/Bcfg2/Server/MultiprocessingCore.py
index 4c304d28c..8031c69d4 100644
--- a/src/lib/Bcfg2/Server/MultiprocessingCore.py
+++ b/src/lib/Bcfg2/Server/MultiprocessingCore.py
@@ -291,6 +291,13 @@ class ChildCore(BaseCore):
         self.metadata_cache.expire(client)
 
     @exposed
+    def RecvProbeData(self, address, _):
+        """ Expire the probe cache for a client """
+        if 'Probes' in self.plugins:
+            client = self.resolve_client(address, metadata=False)
+            self.plugins['Probes'].load_data(client)
+
+    @exposed
     def GetConfig(self, client):
         """ Render the configuration for a client """
         self.logger.debug("%s: Building configuration for %s" %
@@ -389,6 +396,14 @@ class Core(BuiltinCore):
         return BuiltinCore.set_debug(self, address, debug)
 
     @exposed
+    def RecvProbeData(self, address, probedata):
+        rv = BuiltinCore.RecvProbeData(self, address, probedata)
+        # we don't want the children to actually process probe data,
+        # so we don't send the data, just the fact that we got some.
+        self.rpc_q.publish("RecvProbeData", args=[address, None])
+        return rv
+
+    @exposed
     def GetConfig(self, address):
         client = self.resolve_client(address)[0]
         childname = self.children.next()
diff --git a/src/lib/Bcfg2/Server/Plugins/Probes.py b/src/lib/Bcfg2/Server/Plugins/Probes.py
index 407cfc2d4..b58fbf715 100644
--- a/src/lib/Bcfg2/Server/Plugins/Probes.py
+++ b/src/lib/Bcfg2/Server/Plugins/Probes.py
@@ -266,12 +266,14 @@ class Probes(Bcfg2.Server.Plugin.Probing,
             hostname=client.hostname).exclude(
             group__in=self.cgroups[client.hostname]).delete()
 
-    def load_data(self):
+    def load_data(self, client=None):
         """ Load probe data from the appropriate backend (probed.xml
         or the database) """
         if self._use_db:
-            return self._load_data_db()
+            return self._load_data_db(client)
         else:
+            # the XML backend doesn't support loading data for single
+            # clients, so it reloads all data
             return self._load_data_xml()
 
     def _load_data_xml(self):
@@ -296,16 +298,23 @@ class Probes(Bcfg2.Server.Plugin.Probing,
                 elif pdata.tag == 'Group':
                     self.cgroups[client.get('name')].append(pdata.get('name'))
 
-    def _load_data_db(self):
+    def _load_data_db(self, client=None):
         """ Load probe data from the database """
         self.probedata = {}
         self.cgroups = {}
-        for pdata in ProbesDataModel.objects.all():
+        if client is None:
+            probedata = ProbesDataModel.objects.all()
+            groupdata = ProbesGroupsModel.objects.all()
+        else:
+            probedata = ProbesDataModel.objects.filter(hostname=client)
+            groupdata = ProbesGroupsModel.objects.filter(hostname=client)
+
+        for pdata in probedata:
             if pdata.hostname not in self.probedata:
                 self.probedata[pdata.hostname] = ClientProbeDataSet(
                     timestamp=time.mktime(pdata.timestamp.timetuple()))
             self.probedata[pdata.hostname][pdata.probe] = ProbeData(pdata.data)
-        for pgroup in ProbesGroupsModel.objects.all():
+        for pgroup in groupdata:
             if pgroup.hostname not in self.cgroups:
                 self.cgroups[pgroup.hostname] = []
             self.cgroups[pgroup.hostname].append(pgroup.group)