Merge "Fix bug in trigger for initialize_tables()"

congress/datalog/nonrecursive.py

@@ -51,13 +51,18 @@ def initialize_tables(self, tablenames, facts):
             self.rules.clear_table(t)
 
         count = 0
+        extra_tables = set()
+        ignored_facts = 0
         for f in facts:
             if f.table not in cleared_tables:
-                self.rules.clear_table(f.table)
-                cleared_tables.add(f.table)
+                extra_tables.add(f.table)
+                ignored_facts += 1
             self.rules.add_rule(f.table, f)
             count += 1
-
+        if ignored_facts > 0:
+            LOG.error("initialize_tables ignored %d facts for tables "
+                      "%s not included in the list of tablenames %s",
+                      ignored_facts, extra_tables, cleared_tables)
         LOG.info("initialized %d tables with %d facts",
                  len(cleared_tables), count)
 

congress/datasources/nova_driver.py

@@ -16,6 +16,9 @@
 
 from congress.datasources import datasource_driver
 from congress.datasources import datasource_utils
+from congress.openstack.common import log as logging
+
+LOG = logging.getLogger(__name__)
 
 
 def d6service(name, keys, inbox, datapath, args):
@@ -166,4 +169,5 @@ def execute(self, action, action_args):
         # action can be written as a method or an API call.
         # action_agrs can be utilized for distinguishing the two.
         # This is an API call via client:
+        LOG.info("%s:: executing %s on %s", self.name, action, action_args)
         self._execute_api(self.nova_client, action, action_args)

congress/policy_engines/agnostic.py

@@ -179,6 +179,14 @@ def relevant_triggers(self, events):
                 triggers |= self.index[table]
         return triggers
 
+    def _index_string(self):
+        """Build string representation of self.index; useful for debugging."""
+        s = '{'
+        s += ";".join(["%s -> %s" % (key, ",".join(str(x) for x in value))
+                       for key, value in self.index.iteritems()])
+        s += '}'
+        return s
+
     @classmethod
     def triggers_by_table(cls, triggers):
         """Return dictionary from tables to triggers."""
@@ -387,7 +395,25 @@ def initialize_tables(self, tablenames, facts, target=None):
         @facts must be an iterable containing compile.Fact objects.
         """
         target_theory = self.get_target(target)
+        alltables = set([compile.build_tablename(target_theory.name, x)
+                         for x in tablenames])
+        triggers = self.trigger_registry.relevant_triggers(alltables)
+        LOG.info("relevant triggers (init): %s",
+                 ";".join(str(x) for x in triggers))
+        # run queries on relevant triggers *before* applying changes
+        table_triggers = self.trigger_registry.triggers_by_table(triggers)
+        table_data_old = self._compute_table_contents(table_triggers)
+        # actually apply the updates
         target_theory.initialize_tables(tablenames, facts)
+        # rerun the trigger queries to check for changes
+        table_data_new = self._compute_table_contents(table_triggers)
+        # run triggers if tables changed
+        for table, triggers in table_triggers.iteritems():
+            if table_data_old[table] != table_data_new[table]:
+                for trigger in triggers:
+                    trigger.callback(table,
+                                     table_data_old[table],
+                                     table_data_new[table])
 
     def insert(self, formula, target=None):
         """Event handler for arbitrary insertion (rules and facts)."""
@@ -604,7 +630,8 @@ def _update_obj(self, events, theory_string):
         self._maintain_triggers()
         # figure out relevant triggers
         triggers = self.trigger_registry.relevant_triggers(events)
-        LOG.info("relevant triggers: %s", ";".join(str(x) for x in triggers))
+        LOG.info("relevant triggers (update): %s",
+                 ";".join(str(x) for x in triggers))
         # signal trigger registry about graph updates
         self.trigger_registry.update_dependencies(graph_changes)
 
@@ -1430,13 +1457,19 @@ def _maintain_triggers(self):
                     modal='execute')
                 self.execution_triggers[table] = trig
         # remove triggers no longer needed
-        for table in self.execution_triggers:
+        #    Using copy of execution_trigger keys so we can delete inside loop
+        for table in self.execution_triggers.keys():
             LOG.debug("%s:: checking for stale trigger table %s",
                       self.name, table)
             if table not in curr_tables:
                 LOG.debug("removing trigger for table %s", table)
-                self.trigger_registry.unregister(
-                    self.execution_triggers[table])
+                try:
+                    self.trigger_registry.unregister(
+                        self.execution_triggers[table])
+                    del self.execution_triggers[table]
+                except KeyError:
+                    LOG.exception(
+                        "Tried to unregister non-existent trigger: %s", table)
 
     def _execute_table(self, theory, table, old, new):
         # LOG.info("execute_table(theory=%s, table=%s, old=%s, new=%s",

congress/tests/policy_engines/test_agnostic.py

@@ -613,6 +613,18 @@ def test_modal(self):
         run.insert('q(2)')
         self.assertEqual(obj.value, 2)
 
+    def test_initialize(self):
+        obj = self.MyObject()
+        run = agnostic.Runtime()
+        run.debug_mode()
+        run.create_policy('alice')
+        run.register_trigger('p', lambda tbl, old, new:
+                             obj.increment(), 'alice', 'execute')
+        run.insert('execute[p(x)] :- q(x)')
+        self.assertEqual(obj.value, 0)
+        run.initialize_tables(['q'], [compile.Fact('q', [1])], 'alice')
+        self.assertEqual(obj.value, 1)
+
 
 class TestMultipolicyRules(base.TestCase):
     def test_external(self):