From 69e832f36e760e3279468d0615177015f2655077 Mon Sep 17 00:00:00 2001
From: Carlos Goncalves <carlos.goncalves@neclab.eu>
Date: Thu, 18 May 2017 13:50:49 +0000
Subject: [PATCH] WIP: notifier


diff --git a/congress/common/config.py b/congress/common/config.py
index b52f840..44495be 100644
--- a/congress/common/config.py
+++ b/congress/common/config.py
@@ -25,6 +25,7 @@ from oslo_log import log as logging
 from oslo_middleware import cors
 from oslo_policy import opts as policy_opts
 
+from congress.common import notifier
 from congress import version
 
 LOG = logging.getLogger(__name__)
@@ -141,6 +142,8 @@ def init(args, **kwargs):
              version='%%(prog)s %s' % version.version_info.release_string(),
              **kwargs)
 
+    notifier.init()
+
 
 def setup_logging():
     """Sets up logging for the congress package."""
diff --git a/congress/common/notifier.py b/congress/common/notifier.py
new file mode 100644
index 0000000..3aa8087
--- /dev/null
+++ b/congress/common/notifier.py
@@ -0,0 +1,64 @@
+# Copyright (c) 2017 NEC Corporation.  All rights reserved.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import socket
+
+from oslo_config import cfg
+from oslo_log import log as logging
+import oslo_messaging
+
+
+notifier_opts = [
+    cfg.StrOpt('default_publisher_id',
+               default="policy.localhost",
+               help=_("""
+Default publisher_id for outgoing Congress notifications.
+
+This is the value that the notification driver will use to identify
+messages for events originating from the Congress service. Typically,
+this is the hostname of the instance that generated the message.
+
+Possible values:
+    * Any reasonable instance identifier, for example: policy.host1
+
+Related options:
+    * None
+
+""")),
+]
+
+CONF = cfg.CONF
+CONF.register_opts(notifier_opts)
+
+LOG = logging.getLogger(__name__)
+NOTIFIER = None
+
+
+oslo_messaging.set_transport_defaults(control_exchange='congress')
+
+
+def init():
+    global NOTIFIER
+    transport = oslo_messaging.get_notification_transport(CONF)
+    NOTIFIER = oslo_messaging.Notifier(transport,
+                                       'policy',
+                                       driver='messaging',
+                                       topics=['notifications'])
+
+
+def get_notifier(service='policy', host=socket.gethostname(), publisher_id=None):
+    assert NOTIFIER is not None
+    if not publisher_id:
+        publisher_id = "%s.%s" % (service, host)
+    return NOTIFIER.prepare(publisher_id=publisher_id)
diff --git a/congress/policy_engines/agnostic.py b/congress/policy_engines/agnostic.py
index 0753a94..bbd9a52 100644
--- a/congress/policy_engines/agnostic.py
+++ b/congress/policy_engines/agnostic.py
@@ -28,6 +28,7 @@ from oslo_utils import uuidutils
 import six
 from six.moves import range
 
+from congress.common import notifier
 from congress.datalog import base
 from congress.datalog import compile
 from congress.datalog import database as db
@@ -240,6 +241,7 @@ class Runtime (object):
         # rules with errors (because of schema inconsistencies)
         self.error_events = []
         self.synchronizer = None
+        self._notifier = notifier.get_notifier()
 
     ###############################################
     # Persistence layer
@@ -1988,6 +1990,11 @@ class DseRuntime (Runtime, data_service.DataService):
             args = [term.name for term in newlit.arguments]
             LOG.info("%s:: on service %s executing %s on %s",
                      self.name, service, tablename, args)
+            payload = {'service': service,
+                       'tablename': tablename,
+                       'theory': theory,
+                       'args': args}
+            self._notifier.info({}, "policy.engine.execute", payload)
             try:
                 # Note(thread-safety): blocking call
                 self.execute_action(service, tablename, {'positional': args})
-- 
2.7.4

