fjimDMch

1# Copyright (c) 2013 VMware, Inc. All rights reserved.
2#
3#    Licensed under the Apache License, Version 2.0 (the "License"); you may
4#    not use this file except in compliance with the License. You may obtain
5#    a copy of the License at
6#
7#         http://www.apache.org/licenses/LICENSE-2.0
8#
9#    Unless required by applicable law or agreed to in writing, software
10#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
11#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12#    License for the specific language governing permissions and limitations
13#    under the License.
14#
15
16from __future__ import print_function
17from __future__ import division
18from __future__ import absolute_import
19
20import eventlet
21from futurist import periodics
22from oslo_concurrency import lockutils
23from oslo_config import cfg
24from oslo_log import log as logging
25from oslo_utils import uuidutils
26import six
27from six.moves import range
28
29from congress.datalog import base
30from congress.datalog import compile
31from congress.datalog import database as db
32from congress.datalog import materialized
33from congress.datalog import nonrecursive
34from congress.datalog import unify
35from congress.datalog import utility
36from congress.db import db_policy_rules
37from congress.dse2 import data_service
38from congress import exception
39from oslo_messaging import exceptions as messaging_exceptions
40import time
41
42LOG = logging.getLogger(__name__)
43
44
45class ExecutionLogger(object):
46    def __init__(self):
47        self.messages = []
48
49    def debug(self, msg, *args):
50        self.messages.append(msg % args)
51
52    def info(self, msg, *args):
53        self.messages.append(msg % args)
54
55    def warn(self, msg, *args):
56        self.messages.append(msg % args)
57
58    def error(self, msg, *args):
59        self.messages.append(msg % args)
60
61    def critical(self, msg, *args):
62        self.messages.append(msg % args)
63
64    def content(self):
65        return '\n'.join(self.messages)
66
67    def empty(self):
68        self.messages = []
69
70
71def list_to_database(atoms):
72    database = db.Database()
73    for atom in atoms:
74        if atom.is_atom():
75            database.insert(atom)
76    return database
77
78
79def string_to_database(string, theories=None):
80    return list_to_database(compile.parse(
81        string, theories=theories))
82
83
84##############################################################################
85# Runtime
86##############################################################################
87
88class Trigger(object):
89    """A chunk of code that should be run when a table's contents changes."""
90
91    def __init__(self, tablename, policy, callback, modal=None):
92        self.tablename = tablename
93        self.policy = policy
94        self.callback = callback
95        self.modal = modal
96
97    def __str__(self):
98        return "Trigger on table=%s; policy=%s; modal=%s with callback %s." % (
99            self.tablename, self.policy, self.modal, self.callback)
100
101
102class TriggerRegistry(object):
103    """A collection of triggers and algorithms to analyze that collection."""
104
105    def __init__(self, dependency_graph):
106        # graph containing relationships between tables
107        self.dependency_graph = dependency_graph
108
109        # set of triggers that are currently registered
110        self.triggers = set()
111
112        # map from table to triggers relevant to changes for that table
113        self.index = {}
114
115    def register_table(self, tablename, policy, callback, modal=None):
116        """Register CALLBACK to run when TABLENAME changes."""
117        # TODO(thinrichs): either fix dependency graph to differentiate
118        #   between execute[alice:p] and alice:p or reject rules
119        #   in which both occur
120        trigger = Trigger(tablename, policy, callback, modal=modal)
121        self.triggers.add(trigger)
122        self._add_indexes(trigger)
123        LOG.info("registered trigger: %s", trigger)
124        return trigger
125
126    def unregister(self, trigger):
127        """Unregister trigger ID."""
128        self.triggers.remove(trigger)
129        self._delete_indexes(trigger)
130
131    def update_dependencies(self, dependency_graph_changes=None):
132        """Inform registry of changes to the dependency graph.
133
134        Changes are accounted for in self.dependency_graph, but
135        by giving the list of changes we can avoid recomputing
136        all dependencies from scratch.
137        """
138        # TODO(thinrichs): instead of destroying the index and
139        #   recomputing from scratch, look at the changes and
140        #   figure out the delta.
141        self.index = {}
142        for trigger in self.triggers:
143            self._add_indexes(trigger)
144
145    def _add_indexes(self, trigger):
146        full_table = compile.Tablename.build_service_table(
147            trigger.policy, trigger.tablename)
148        deps = self.dependency_graph.dependencies(full_table)
149        if deps is None:
150            deps = set([full_table])
151        for table in deps:
152            if table in self.index:
153                self.index[table].add(trigger)
154            else:
155                self.index[table] = set([trigger])
156
157    def _delete_indexes(self, trigger):
158        full_table = compile.Tablename.build_service_table(
159            trigger.policy, trigger.tablename)
160        deps = self.dependency_graph.dependencies(full_table)
161        if deps is None:
162            deps = set([full_table])
163        for table in deps:
164            self.index[table].discard(trigger)
165
166    def relevant_triggers(self, events):
167        """Return the set of triggers that are relevant to the EVENTS.
168
169        Each EVENT may either be a compile.Event or a tablename.
170        """
171        table_changes = set()
172        for event in events:
173            if isinstance(event, compile.Event):
174                if compile.is_rule(event.formula):
175                    table_changes |= set(
176                        [lit.table.global_tablename(event.target)
177                         for lit in event.formula.heads])
178                else:
179                    table_changes.add(
180                        event.formula.table.global_tablename(event.target))
181            elif isinstance(event, six.string_types):
182                table_changes.add(event)
183        triggers = set()
184        for table in table_changes:
185            if table in self.index:
186                triggers |= self.index[table]
187        return triggers
188
189    def _index_string(self):
190        """Build string representation of self.index; useful for debugging."""
191        s = '{'
192        s += ";".join(["%s -> %s" % (key, ",".join(str(x) for x in value))
193                       for key, value in self.index.items()])
194        s += '}'
195        return s
196
197    @classmethod
198    def triggers_by_table(cls, triggers):
199        """Return dictionary from tables to triggers."""
200        d = {}
201        for trigger in triggers:
202            table = (trigger.tablename, trigger.policy, trigger.modal)
203            if table not in d:
204                d[table] = [trigger]
205            else:
206                d[table].append(trigger)
207        return d
208
209
210class Runtime (object):
211    """Runtime for the Congress policy language.
212
213    Only have one instantiation in practice, but using a
214    class is natural and useful for testing.
215    """
216
217    DEFAULT_THEORY = 'classification'
218    ACTION_THEORY = 'action'
219
220    def __init__(self):
221        # tracer object
222        self.tracer = base.Tracer()
223        # record execution
224        self.logger = ExecutionLogger()
225        # collection of theories
226        self.theory = {}
227        # collection of builtin theories
228        self.builtin_policy_names = set()
229        # dependency graph for all theories
230        self.global_dependency_graph = (
231            compile.RuleDependencyGraph())
232        # triggers
233        self.trigger_registry = TriggerRegistry(self.global_dependency_graph)
234        # execution triggers
235        self.execution_triggers = {}
236        # disabled rules
237        self.disabled_events = []
238        # rules with errors (because of schema inconsistencies)
239        self.error_events = []
240
241    # synchronizer is used in both Runtime and DseRuntime, so adding this as
242    # part of parent class.
243    # TODO(ramineni): For better deisgn, move this to differnt class and
244    # perform the operations
245    @lockutils.synchronized('congress_synchronize_policies')
246    def synchronize_policies(self, policy_name=None):
247        # Read policies from DB.
248        configured_policies = [{'id': p.id,
249                                'name': p.name,
250                                'abbr': p.abbreviation,
251                                'desc': p.description,
252                                'owner': p.owner,
253                                'kind': p.kind}
254                               for p in db_policy_rules.get_policies()]
255
256        if policy_name:
257            # synchronize only one policy if doesn't exist
258            p = db_policy_rules.get_policy_by_name(policy_name)
259            if p and policy_name not in self.policy_names():
260                self.create_policy(p.name, id_=p.id, abbr=p.abbreviation,
261                                   kind=p.kind, desc=p.description,
262                                   owner=p.owner)
263                LOG.debug("synchronizer, policy added %s", policy_name)
264            elif p is None and policy_name in self.policy_names():
265                self.delete_policy(policy_name)
266                LOG.debug("synchronizer, policy deleted %s", policy_name)
267            return
268
269        # Read policies from engine
270        policies = [self.policy_object(n) for n in self.policy_names()]
271        active_policies = []
272        for policy in policies:
273            active_policies.append({'id': policy.id,
274                                    'name': policy.name,
275                                    'abbr': policy.abbr,
276                                    'desc': policy.desc,
277                                    'owner': policy.owner,
278                                    'kind': policy.kind})
279
280        added = 0
281        removed = 0
282        datasource_policies = []
283        for p in active_policies:
284            if (p['kind'] == base.DATASOURCE_POLICY_TYPE):
285                datasource_policies.append(p['name'])
286                continue
287
288            if (p['kind'] != base.DATASOURCE_POLICY_TYPE and
289                    p not in configured_policies):
290                LOG.debug("removing policy %s", str(p))
291                self.delete_policy(p['id'])
292                removed = removed + 1
293
294        for p in configured_policies:
295            if p not in active_policies:
296                LOG.debug("adding policy %s", str(p))
297                self.create_policy(p['name'], id_=p['id'], abbr=p['abbr'],
298                                   kind=p['kind'], desc=p['desc'],
299                                   owner=p['owner'])
300                added = added + 1
301
302        # Synchronize datasource policies with datasources in DB
303        db_datasources = self.node.get_datasources()
304        for p in db_datasources:
305            ds_name = p['name']
306            if ds_name not in datasource_policies:
307                # datasource not registered with PE, sync it
308
309                if not self.node.is_valid_service(ds_name):
310                    # datasource service is not yet up, sync in next iter
311                    continue
312
313                # Get the datasource schema to sync the schema with PE
314                schema = self.rpc(ds_name, 'get_datasource_schema',
315                                  {'source_id': p['name']})
316                self.initialize_datasource(ds_name, schema)
317                LOG.debug("Initialized datasource %s on node %s", ds_name,
318                          self.node.node_id)
319                added = added + 1
320            else:
321                datasource_policies.remove(p['name'])
322
323        # Remove datasource policies in PE that doesn't exist in DB
324        for p in datasource_policies:
325            self.delete_policy(p)
326            removed = removed + 1
327
328        LOG.debug("synchronize_policies, added %d removed %d ", added, removed)
329
330    ###############################################
331    # Persistence layer
332    ###############################################
333    # Note(thread-safety): blocking function
334    def persistent_create_policy(self, name, id_=None, abbr=None, kind=None,
335                                 desc=None):
336        # validation for name
337        try:
338            self.parse("%s() :- true()" % name)
339        except exception.PolicyException:
340            raise exception.PolicyException(
341                "Policy name %s is not a valid tablename" % name)
342
343        # Create policy object in policy engine.
344        if id_ is None:
345            id_ = str(uuidutils.generate_uuid())
346        policy_obj = self.create_policy(
347            name=name, abbr=abbr, kind=kind, id_=id_, desc=desc, owner='user')
348
349        # save policy to database
350        if desc is None:
351            desc = ''
352        obj = {'id': policy_obj.id,
353               'name': policy_obj.name,
354               'owner_id': 'user',
355               'description': desc,
356               'abbreviation': policy_obj.abbr,
357               'kind': policy_obj.kind}
358        try:
359            # Note(thread-safety): blocking function
360            db_policy_rules.add_policy(obj['id'],
361                                       obj['name'],
362                                       obj['abbreviation'],
363                                       obj['description'],
364                                       obj['owner_id'],
365                                       obj['kind'])
366            self.synchronize_policies(policy_name=obj['name'])
367        except Exception as e:
368            policy_name = policy_obj.name
369            msg = "Error thrown while adding policy %s into DB." % policy_name
370            LOG.exception(msg)
371            LOG.exception(e)
372            self.delete_policy(policy_name)
373            raise exception.PolicyException(msg)
374        return obj
375
376    # Note(thread-safety): blocking function
377    def persistent_delete_policy(self, name_or_id):
378        # Note(thread-safety): blocking call
379        db_object = db_policy_rules.get_policy(name_or_id)
380        if db_object['name'] in [self.DEFAULT_THEORY, self.ACTION_THEORY]:
381            raise KeyError("Cannot delete system-maintained policy %s" %
382                           db_object['name'])
383        # delete policy from memory and from database
384        db_policy_rules.delete_policy(db_object['id'])
385        self.synchronize_policies(policy_name=db_object['name'])
386        return db_object.to_dict()
387
388    # Note(thread-safety): blocking function
389    def persistent_get_policies(self):
390        return [p.to_dict() for p in db_policy_rules.get_policies()]
391
392    # Note(thread-safety): blocking function
393    def persistent_get_policy(self, id_):
394        # Note(thread-safety): blocking call
395        try:
396            policy = db_policy_rules.get_policy(id_)
397            return policy.to_dict()
398        except KeyError:
399            raise exception.NotFound(
400                'No policy found with name or id %s' % id_)
401
402    # Note(thread-safety): blocking function
403    def persistent_get_rule(self, id_, policy_name):
404        """Return data for rule with id_ in policy_name."""
405        # Check if policy exists, else raise error
406        self.assert_policy_exists(policy_name)
407        # Note(thread-safety): blocking call
408        rule = db_policy_rules.get_policy_rule(id_, policy_name)
409        if rule is None:
410            return
411        return rule.to_dict()
412
413    # Note(thread-safety): blocking function
414    def persistent_get_rules(self, policy_name):
415        """Return data for all rules in policy_name."""
416        # Check if policy exists, else raise error
417        self.assert_policy_exists(policy_name)
418        # Note(thread-safety): blocking call
419        rules = db_policy_rules.get_policy_rules(policy_name)
420        return [rule.to_dict() for rule in rules]
421
422    # Note(thread-safety): blocking function
423    def persistent_insert_rule(self, policy_name, str_rule, rule_name,
424                               comment):
425        """Insert and persists rule into policy_name."""
426        # Reject rules inserted into non-persisted policies
427        # (i.e. datasource policies)
428        # Note(thread-safety): blocking call
429        policy_name = db_policy_rules.policy_name(policy_name)
430        # call synchronizer to make sure policy is synchronized in memory
431        self.synchronize_policies(policy_name=policy_name)
432        # Note(thread-safety): blocking call
433        policies = db_policy_rules.get_policies()
434        persisted_policies = set([p.name for p in policies])
435        if policy_name not in persisted_policies:
436            if policy_name in self.theory:
437                LOG.debug(
438                    "insert_persisted_rule error: rule not permitted for "
439                    "policy %s", policy_name)
440                raise exception.PolicyRuntimeException(
441                    name='rule_not_permitted')
442
443        id_ = uuidutils.generate_uuid()
444        try:
445            rule = self.parse(str_rule)
446        except exception.PolicyException as e:
447            # TODO(thinrichs): change compiler to provide these error_code
448            #   names directly.
449            raise exception.PolicyException(str(e), name='rule_syntax')
450
451        if len(rule) == 1:
452            rule = rule[0]
453        else:
454            msg = ("Received multiple rules: " +
455                   "; ".join(str(x) for x in rule))
456            raise exception.PolicyRuntimeException(msg, name='multiple_rules')
457
458        rule.set_id(id_)
459        rule.set_name(rule_name)
460        rule.set_comment(comment or "")
461        rule.set_original_str(str_rule)
462        changes = self._safe_process_policy_update(
463            rule, policy_name, persistent=True)
464        # save rule to database if change actually happened.
465        # Note: change produced may not be equivalent to original rule because
466        #    of column-reference elimination.
467        if len(changes) > 0:
468            d = {'rule': rule.pretty_str(),
469                 'id': str(rule.id),
470                 'comment': rule.comment,
471                 'name': rule.name}
472            try:
473                # Note(thread-safety): blocking call
474                db_policy_rules.add_policy_rule(
475                    d['id'], policy_name, str_rule, d['comment'],
476                    rule_name=d['name'])
477                return (d['id'], d)
478            except Exception as db_exception:
479                try:
480                    self._safe_process_policy_update(
481                        rule, policy_name, insert=False)
482                    raise exception.PolicyRuntimeException(
483                        "Error while writing to DB: %s."
484                        % str(db_exception))
485                except Exception as change_exception:
486                    raise exception.PolicyRuntimeException(
487                        "Error thrown during recovery from DB error. "
488                        "Inconsistent state.  DB error: %s.  "
489                        "New error: %s." % (str(db_exception),
490                                            str(change_exception)))
491
492        # change not accepted means it was already there
493        raise exception.PolicyRuntimeException(
494            name='rule_already_exists')
495
496    # Note(thread-safety): blocking function
497    def persistent_delete_rule(self, id_, policy_name_or_id):
498        # Note(thread-safety): blocking call
499        policy_name = db_policy_rules.policy_name(policy_name_or_id)
500        # Note(thread-safety): blocking call
501        item = self.persistent_get_rule(id_, policy_name)
502        if item is None:
503            raise exception.PolicyRuntimeException(
504                name='rule_not_exists',
505                data='ID: %s, policy_name: %s' % (id_, policy_name))
506        rule = self.parse1(item['rule'])
507        self._safe_process_policy_update(rule, policy_name, insert=False)
508        # Note(thread-safety): blocking call
509        db_policy_rules.delete_policy_rule(id_)
510        return item
511
512    # Note(thread-safety): blocking function
513    def persistent_load_policies(self):
514        """Load policies from database."""
515        # Note(thread-safety): blocking call
516        for policy in db_policy_rules.get_policies():
517            # Note(thread-safety): blocking call
518            self.create_policy(policy.name, abbr=policy.abbreviation,
519                               kind=policy.kind, id_=policy.id,
520                               desc=policy.description, owner=policy.owner)
521
522    # Note(thread-safety): blocking function
523    def persistent_load_rules(self):
524        """Load all rules from the database."""
525        # Note(thread-safety): blocking call
526        rules = db_policy_rules.get_policy_rules()
527        for rule in rules:
528            parsed_rule = self.parse1(rule.rule)
529            parsed_rule.set_id(rule.id)
530            parsed_rule.set_name(rule.name)
531            parsed_rule.set_comment(rule.comment)
532            parsed_rule.set_original_str(rule.rule)
533            self._safe_process_policy_update(
534                parsed_rule,
535                rule.policy_name)
536
537    def _safe_process_policy_update(self, parsed_rule, policy_name,
538                                    insert=True, persistent=False):
539        if policy_name not in self.theory:
540            raise exception.PolicyRuntimeException(
541                'Policy ID %s does not exist' % policy_name,
542                name='policy_not_exist')
543        event = compile.Event(
544            formula=parsed_rule,
545            insert=insert,
546            target=policy_name)
547        (permitted, changes) = self.process_policy_update(
548            [event], persistent=persistent)
549        if not permitted:
550            raise exception.PolicyException(
551                ";".join([str(x) for x in changes]),
552                name='rule_syntax')
553        return changes
554
555    def process_policy_update(self, events, persistent=False):
556        LOG.debug("process_policy_update %s" % events)
557        # body_only so that we don't subscribe to tables in the head
558        result = self.update(events, persistent=persistent)
559        return result
560
561    ##########################
562    # Non-persistence layer
563    ##########################
564
565    def create_policy(self, name, abbr=None, kind=None, id_=None,
566                      desc=None, owner=None):
567        """Create a new policy and add it to the runtime.
568
569        ABBR is a shortened version of NAME that appears in
570        traces.  KIND is the name of the datastructure used to
571        represent a policy.
572        """
573        if not isinstance(name, six.string_types):
574            raise KeyError("Policy name %s must be a string" % name)
575        if name in self.theory:
576            raise KeyError("Policy with name %s already exists" % name)
577        if not isinstance(abbr, six.string_types):
578            abbr = name[0:5]
579        LOG.debug("Creating policy <%s> with abbr <%s> and kind <%s>",
580                  name, abbr, kind)
581        if kind is None:
582            kind = base.NONRECURSIVE_POLICY_TYPE
583        else:
584            kind = kind.lower()
585        if kind == base.NONRECURSIVE_POLICY_TYPE:
586            PolicyClass = nonrecursive.NonrecursiveRuleTheory
587        elif kind == base.ACTION_POLICY_TYPE:
588            PolicyClass = nonrecursive.ActionTheory
589        elif kind == base.DATABASE_POLICY_TYPE:
590            PolicyClass = db.Database
591        elif kind == base.MATERIALIZED_POLICY_TYPE:
592            PolicyClass = materialized.MaterializedViewTheory
593        elif kind == base.DATASOURCE_POLICY_TYPE:
594            PolicyClass = nonrecursive.DatasourcePolicyTheory
595        else:
596            raise exception.PolicyException(
597                "Unknown kind of policy: %s" % kind)
598        policy_obj = PolicyClass(name=name, abbr=abbr, theories=self.theory,
599                                 desc=desc, owner=owner)
600        policy_obj.set_id(id_)
601        policy_obj.set_tracer(self.tracer)
602        self.theory[name] = policy_obj
603        LOG.debug("Created policy <%s> with abbr <%s> and kind <%s>",
604                  policy_obj.name, policy_obj.abbr, policy_obj.kind)
605        return policy_obj
606
607    def initialize_datasource(self, name, schema):
608        """Initializes datasource by creating policy and setting schema. """
609        try:
610            self.create_policy(name, kind=base.DATASOURCE_POLICY_TYPE)
611        except KeyError:
612            raise exception.DatasourceNameInUse(value=name)
613        try:
614            self.set_schema(name, schema)
615        except Exception:
616            self.delete_policy(name)
617            raise exception.DatasourceCreationError(value=name)
618
619    def delete_policy(self, name_or_id, disallow_dangling_refs=False):
620        """Deletes policy with name NAME or throws KeyError or DanglingRefs."""
621        LOG.info("Deleting policy named %s", name_or_id)
622        name = self._find_policy_name(name_or_id)
623        if disallow_dangling_refs:
624            refs = self._references_to_policy(name)
625            if refs:
626                refmsg = ";".join("%s: %s" % (policy, rule)
627                                  for policy, rule in refs)
628                raise exception.DanglingReference(
629                    "Cannot delete %s because it would leave dangling "
630                    "references: %s" % (name, refmsg))
631        # delete the rules explicitly so cross-theory state is properly
632        #   updated
633        events = [compile.Event(formula=rule, insert=False, target=name)
634                  for rule in self.theory[name].content()]
635        permitted, errs = self.update(events)
636        if not permitted:
637            # This shouldn't happen
638            msg = ";".join(str(x) for x in errs)
639            LOG.exception("%s:: failed to empty theory %s: %s",
640                          self.name, name, msg)
641            raise exception.PolicyException("Policy %s could not be deleted "
642                                            "since rules could not all be "
643                                            "deleted: %s" % (name, msg))
644        # delete disabled rules
645        self.disabled_events = [event for event in self.disabled_events
646                                if event.target != name]
647        # actually delete the theory
648        del self.theory[name]
649
650    def rename_policy(self, oldname, newname):
651        """Renames policy OLDNAME to NEWNAME or raises KeyError."""
652        if newname in self.theory:
653            raise KeyError('Cannot rename %s to %s: %s already exists' %
654                           (oldname, newname, newname))
655        try:
656            self.theory[newname] = self.theory[oldname]
657            del self.theory[oldname]
658        except KeyError:
659            raise KeyError('Cannot rename %s to %s: %s does not exist' %
660                           (oldname, newname, oldname))
661
662    # TODO(thinrichs): make Runtime act like a dictionary so that we
663    #   can iterate over policy names (keys), check if a policy exists, etc.
664    def assert_policy_exists(self, policy_name):
665        """Checks if policy exists or not.
666
667        :param policy_name: policy name
668        :returns: True, if policy exists
669        :raises: PolicyRuntimeException, if policy doesn't exist.
670        """
671        if policy_name not in self.theory:
672            raise exception.PolicyRuntimeException(
673                'Policy ID %s does not exist' % policy_name,
674                name='policy_not_exist')
675        return True
676
677    def policy_names(self):
678        """Returns list of policy names."""
679        return list(self.theory.keys())
680
681    def policy_object(self, name=None, id=None):
682        """Return policy by given name.  Raises KeyError if does not exist."""
683        assert name or id
684        if name:
685            try:
686                if not id or str(self.theory[name].id) == str(id):
687                    return self.theory[name]
688            except KeyError:
689                raise KeyError("Policy with name %s and id %s does not "
690                               "exist" % (name, str(id)))
691        elif id:
692            for n in self.policy_names():
693                if str(self.theory[n].id) == str(id):
694                    return self.theory[n]
695            raise KeyError("Policy with name %s and id %s does not "
696                           "exist" % (name, str(id)))
697
698    def policy_type(self, name):
699        """Return type of policy NAME.  Throws KeyError if does not exist."""
700        return self.policy_object(name).kind
701
702    def set_schema(self, name, schema, complete=False):
703        """Set the schema for module NAME to be SCHEMA."""
704        # TODO(thinrichs): handle the case of a schema being UPDATED,
705        #   not just being set for the first time
706        if name not in self.theory:
707            raise exception.CongressException(
708                "Cannot set policy for %s because it has not been created" %
709                name)
710        if self.theory[name].schema and len(self.theory[name].schema) > 0:
711            raise exception.CongressException(
712                "Schema for %s already set" % name)
713        self.theory[name].schema = compile.Schema(schema, complete=complete)
714        enabled, disabled, errs = self._process_limbo_events(
715            self.disabled_events)
716        self.disabled_events = disabled
717        self.error_events.extend(errs)
718        for event in enabled:
719            permitted, errors = self._update_obj_datalog([event])
720            if not permitted:
721                self.error_events.append((event, errors))
722
723    def _create_status_dict(self, target, keys):
724        result = {}
725
726        for k in keys:
727            attr = getattr(target, k, None)
728            if attr is not None:
729                result[k] = attr
730
731        return result
732
733    def get_status(self, source_id, params):
734        try:
735            if source_id in self.policy_names():
736                target = self.policy_object(name=source_id)
737            else:
738                target = self.policy_object(id=source_id)
739            keys = ['name', 'id']
740
741            if 'rule_id' in params:
742                target = target.get_rule(str(params['rule_id']))
743                keys.extend(['comment', 'original_str'])
744
745        except Exception as e:
746            LOG.exception(e)
747            raise exception.NotFound(str(e))
748
749        return self._create_status_dict(target, keys)
750
751    def select(self, query, target=None, trace=False):
752        """Event handler for arbitrary queries.
753
754        Returns the set of all instantiated QUERY that are true.
755        """
756        if isinstance(query, six.string_types):
757            return self._select_string(query, self.get_target(target), trace)
758        elif isinstance(query, tuple):
759            return self._select_tuple(query, self.get_target(target), trace)
760        else:
761            return self._select_obj(query, self.get_target(target), trace)
762
763    def initialize_tables(self, tablenames, facts, target=None):
764        """Event handler for (re)initializing a collection of tables
765
766        @facts must be an iterable containing compile.Fact objects.
767        """
768        target_theory = self.get_target(target)
769        alltables = set([compile.Tablename.build_service_table(
770                         target_theory.name, x)
771                         for x in tablenames])
772        triggers = self.trigger_registry.relevant_triggers(alltables)
773        LOG.info("relevant triggers (init): %s",
774                 ";".join(str(x) for x in triggers))
775        # run queries on relevant triggers *before* applying changes
776        table_triggers = self.trigger_registry.triggers_by_table(triggers)
777        table_data_old = self._compute_table_contents(table_triggers)
778        # actually apply the updates
779        target_theory.initialize_tables(tablenames, facts)
780        # rerun the trigger queries to check for changes
781        table_data_new = self._compute_table_contents(table_triggers)
782        # run triggers if tables changed
783        for table, triggers in table_triggers.items():
784            if table_data_old[table] != table_data_new[table]:
785                for trigger in triggers:
786                    trigger.callback(table,
787                                     table_data_old[table],
788                                     table_data_new[table])
789
790    def insert(self, formula, target=None):
791        """Event handler for arbitrary insertion (rules and facts)."""
792        if isinstance(formula, six.string_types):
793            return self._insert_string(formula, target)
794        elif isinstance(formula, tuple):
795            return self._insert_tuple(formula, target)
796        else:
797            return self._insert_obj(formula, target)
798
799    def delete(self, formula, target=None):
800        """Event handler for arbitrary deletion (rules and facts)."""
801        if isinstance(formula, six.string_types):
802            return self._delete_string(formula, target)
803        elif isinstance(formula, tuple):
804            return self._delete_tuple(formula, target)
805        else:
806            return self._delete_obj(formula, target)
807
808    def update(self, sequence, target=None, persistent=False):
809        """Event handler for applying an arbitrary sequence of insert/deletes.
810
811        If TARGET is supplied, it overrides the targets in SEQUENCE.
812        """
813        if isinstance(sequence, six.string_types):
814            return self._update_string(sequence, target, persistent)
815        else:
816            return self._update_obj(sequence, target, persistent)
817
818    def policy(self, target=None):
819        """Event handler for querying policy."""
820        target = self.get_target(target)
821        if target is None:
822            return ""
823        return " ".join(str(p) for p in target.policy())
824
825    def content(self, target=None):
826        """Event handler for querying content()."""
827        target = self.get_target(target)
828        if target is None:
829            return ""
830        return " ".join(str(p) for p in target.content())
831
832    def simulate(self, query, theory, sequence, action_theory, delta=False,
833                 trace=False, as_list=False):
834        """Event handler for simulation.
835
836        :param query is a string/object to query after
837        :param theory is the policy to query
838        :param sequence is a string/iter of updates to state/policy or actions
839        :param action_theory is the policy that contains action descriptions
840        :param delta indicates whether to return *changes* to query caused by
841               sequence
842        :param trace indicates whether to include a string description of the
843               implementation.  When True causes the return value to be the
844               tuple (result, trace).
845        :param as_list controls whether the result is forced to be a list of
846               answers
847        Returns a list of instances of query.  If query/sequence are strings
848        the query instance list is a single string (unless as_list is True
849        in which case the query instance list is a list of strings).  If
850        query/sequence are objects then the query instance list is a list
851        of objects.
852
853        The computation of a query given an action sequence. That sequence
854        can include updates to atoms, updates to rules, and action
855        invocations.  Returns a collection of Literals (as a string if the
856        query and sequence are strings or as a Python collection otherwise).
857        If delta is True, the return is a collection of Literals where
858        each tablename ends with either + or - to indicate whether
859        that fact was added or deleted.
860        Example atom update: q+(1) or q-(1)
861        Example rule update: p+(x) :- q(x) or p-(x) :- q(x)
862        Example action invocation:
863           create_network(17), options:value(17, "name", "net1") :- true
864        """
865        assert self.get_target(theory) is not None, "Theory must be known"
866        assert self.get_target(action_theory) is not None, (
867            "Action theory must be known")
868        if (isinstance(query, six.string_types) and
869                isinstance(sequence, six.string_types)):
870            return self._simulate_string(query, theory, sequence,
871                                         action_theory, delta, trace, as_list)
872        else:
873            return self._simulate_obj(query, theory, sequence, action_theory,
874                                      delta, trace)
875
876    def get_tablename(self, source_id, table_id):
877        tables = self.get_tablenames(source_id)
878        # when the policy doesn't have any rule 'tables' is set([])
879        # when the policy doesn't exist 'tables' is None
880        if tables and table_id in tables:
881            return table_id
882
883    def get_tablenames(self, source_id):
884        if source_id in self.theory.keys():
885            return self.tablenames(theory_name=source_id, include_modal=False)
886
887    def get_row_data(self, table_id, source_id, trace=False):
888        # source_id is the policy name.  But it needs to stay 'source_id'
889        #  since RPC calls invoke by the name of the argument, and we're
890        #  currently assuming the implementations of get_row_data in
891        #  the policy engine, datasources, and datasource manager all
892        #  use the same argument names.
893        policy_name = source_id
894        tablename = self.get_tablename(policy_name, table_id)
895        if not tablename:
896            raise exception.NotFound("table '%s' doesn't exist" % table_id)
897
898        queries = self.table_contents_queries(tablename, policy_name)
899        if queries is None:
900            m = "Known table but unknown arity for '%s' in policy '%s'" % (
901                tablename, policy_name)
902            LOG.error(m)
903            raise exception.CongressException(m)
904
905        gen_trace = None
906        query = self.parse1(queries[0])
907        # LOG.debug("query: %s", query)
908        result = self.select(query, target=policy_name,
909                             trace=trace)
910        if trace:
911            literals = result[0]
912            gen_trace = result[1]
913        else:
914            literals = result
915        # should NOT need to convert to set -- see bug 1344466
916        literals = frozenset(literals)
917        # LOG.info("results: %s", '\n'.join(str(x) for x in literals))
918        results = []
919        for lit in literals:
920            d = {}
921            d['data'] = [arg.name for arg in lit.arguments]
922            results.append(d)
923
924        if trace:
925            return results, gen_trace
926        else:
927            return results
928
929    def tablenames(self, body_only=False, include_builtin=False,
930                   theory_name=None, include_modal=True):
931        """Return tablenames occurring in some theory."""
932        tables = set()
933
934        if theory_name:
935            th = self.theory.get(theory_name, None)
936            if th:
937                tables |= set(th.tablenames(body_only=body_only,
938                                            include_builtin=include_builtin,
939                                            include_modal=include_modal))
940            return tables
941
942        for th in self.theory.values():
943            tables |= set(th.tablenames(body_only=body_only,
944                                        include_builtin=include_builtin,
945                                        include_modal=include_modal))
946        return tables
947
948    def reserved_tablename(self, name):
949        return name.startswith('___')
950
951    def table_contents_queries(self, tablename, policy, modal=None):
952        """Return list of queries yielding contents of TABLENAME in POLICY."""
953        # TODO(thinrichs): Handle case of multiple arities.  Connect to API.
954        arity = self.arity(tablename, policy, modal)
955        if arity is None:
956            return
957        args = ["x" + str(i) for i in range(0, arity)]
958        atom = tablename + "(" + ",".join(args) + ")"
959        if modal is None:
960            return [atom]
961        else:
962            return [modal + "[" + atom + "]"]
963
964    def register_trigger(self, tablename, callback, policy=None, modal=None):
965        """Register CALLBACK to run when table TABLENAME changes."""
966        # calling self.get_target_name to check if policy actually exists
967        #   and to resolve None to a policy name
968        return self.trigger_registry.register_table(
969            tablename, self.get_target_name(policy), callback, modal=modal)
970
971    def unregister_trigger(self, trigger):
972        """Unregister CALLBACK for table TABLENAME."""
973        return self.trigger_registry.unregister(trigger)
974
975    def arity(self, table, theory, modal=None):
976        """Return number of columns for TABLE in THEORY.
977
978        TABLE can include the policy name. <policy>:<table>
979        THEORY is the name of the theory we are asking.
980        MODAL is the value of the modal, if any.
981        """
982        arity = self.get_target(theory).arity(table, modal)
983        if arity is not None:
984            return arity
985        policy, tablename = compile.Tablename.parse_service_table(table)
986        if policy not in self.theory:
987            return
988        return self.theory[policy].arity(tablename, modal)
989
990    def find_subpolicy(self, required_tables, prohibited_tables,
991                       output_tables, target=None):
992        """Return a subset of rules in @theory.
993
994        @required_tables is the set of tablenames that a rule must depend on.
995        @prohibited_tables is the set of tablenames that a rule must
996        NOT depend on.
997        @output_tables is the set of tablenames that all rules must support.
998        """
999        target = self.get_target(target)
1000        if target is None:
1001            return
1002        subpolicy = compile.find_subpolicy(
1003            target.content(),
1004            required_tables,
1005            prohibited_tables,
1006            output_tables)
1007        return " ".join(str(p) for p in subpolicy)
1008
1009    ##########################################
1010    # Implementation of Non-persistence layer
1011    ##########################################
1012    # Arguments that are strings are suffixed with _string.
1013    # All other arguments are instances of Theory, Literal, etc.
1014
1015    ###################################
1016    # Implementation: updates
1017
1018    # insert: convenience wrapper around Update
1019    def _insert_string(self, policy_string, theory_string):
1020        policy = self.parse(policy_string)
1021        return self._update_obj(
1022            [compile.Event(formula=x, insert=True, target=theory_string)
1023             for x in policy],
1024            theory_string)
1025
1026    def _insert_tuple(self, iter, theory_string):
1027        return self._insert_obj(compile.Literal.create_from_iter(iter),
1028                                theory_string)
1029
1030    def _insert_obj(self, formula, theory_string):
1031        return self._update_obj([compile.Event(formula=formula, insert=True,
1032                                               target=theory_string)],
1033                                theory_string)
1034
1035    # delete: convenience wrapper around Update
1036    def _delete_string(self, policy_string, theory_string):
1037        policy = self.parse(policy_string)
1038        return self._update_obj(
1039            [compile.Event(formula=x, insert=False, target=theory_string)
1040             for x in policy],
1041            theory_string)
1042
1043    def _delete_tuple(self, iter, theory_string):
1044        return self._delete_obj(compile.Literal.create_from_iter(iter),
1045                                theory_string)
1046
1047    def _delete_obj(self, formula, theory_string):
1048        return self._update_obj([compile.Event(formula=formula, insert=False,
1049                                               target=theory_string)],
1050                                theory_string)
1051
1052    # update
1053    def _update_string(self, events_string, theory_string, persistent=False):
1054        assert False, "Not yet implemented--need parser to read events"
1055
1056    def _update_obj(self, events, theory_string, persistent=False):
1057        """Apply events.
1058
1059        Checks if applying EVENTS is permitted and if not
1060        returns a list of errors.  If it is permitted, it
1061        applies it and then returns a list of changes.
1062        In both cases, the return is a 2-tuple (if-permitted, list).
1063        Note: All event.target fields are the NAMES of theories, not
1064        theory objects.  theory_string is the default theory.
1065        """
1066        errors = []
1067        # resolve event targets and check that they actually exist
1068        for event in events:
1069            if event.target is None:
1070                event.target = theory_string
1071            try:
1072                event.target = self.get_target_name(event.target)
1073            except exception.PolicyException as e:
1074                errors.append(e)
1075        if len(errors) > 0:
1076            return (False, errors)
1077        # eliminate column refs where possible
1078        enabled, disabled, errs = self._process_limbo_events(
1079            events, persistent)
1080        for err in errs:
1081            errors.extend(err[1])
1082        if len(errors) > 0:
1083            return (False, errors)
1084        # continue updating and if successful disable the rest
1085        permitted, extra = self._update_obj_datalog(enabled)
1086        if not permitted:
1087            return permitted, extra
1088        self._disable_events(disabled)
1089        return (True, extra)
1090
1091    def _disable_events(self, events):
1092        """Take collection of insert events and disable them.
1093
1094        Assume that events.theory is an object.
1095        """
1096        self.disabled_events.extend(events)
1097
1098    def _process_limbo_events(self, events, persistent=False):
1099        """Assume that events.theory is an object.
1100
1101        Return (<enabled>, <disabled>, <errors>)
1102        where <errors> is a list of (event, err-list).
1103        """
1104        disabled = []
1105        enabled = []
1106        errors = []
1107        for event in events:
1108            errs = compile.check_schema_consistency(
1109                event.formula, self.theory, event.target)
1110            if len(errs) > 0:
1111                errors.append((event, errs))
1112                continue
1113            try:
1114                oldformula = event.formula
1115                event.formula = oldformula.eliminate_column_references(
1116                    self.theory, default_theory=event.target)
1117                # doesn't copy over ID since it creates a new one
1118                event.formula.set_id(oldformula.id)
1119                enabled.append(event)
1120            except exception.IncompleteSchemaException as e:
1121                if persistent:
1122                    # FIXME(ekcs): inconsistent behavior?
1123                    # persistent_insert with 'unknown:p(x)' allowed but
1124                    # 'unknown:p(colname=x)' disallowed
1125                    raise exception.PolicyException(str(e), name='rule_syntax')
1126                else:
1127                    disabled.append(event)
1128            except exception.PolicyException as e:
1129                errors.append((event, [e]))
1130        return enabled, disabled, errors
1131
1132    def _update_obj_datalog(self, events):
1133        """Do the updating.
1134
1135        Checks if applying EVENTS is permitted and if not
1136        returns a list of errors.  If it is permitted, it
1137        applies it and then returns a list of changes.
1138        In both cases, the return is a 2-tuple (if-permitted, list).
1139        Note: All event.target fields are the NAMES of theories, not
1140        theory objects, and all event.formula fields have
1141        had all column references removed.
1142        """
1143        # TODO(thinrichs): look into whether we can move the bulk of the
1144        # trigger code into Theory, esp. so that MaterializedViewTheory
1145        # can implement it more efficiently.
1146        self.table_log(None, "Updating with %s", utility.iterstr(events))
1147        errors = []
1148        # eliminate noop events
1149        events = self._actual_events(events)
1150        if not len(events):
1151            return (True, [])
1152        # check that the updates would not cause an error
1153        by_theory = self._group_events_by_target(events)
1154        for th, th_events in by_theory.items():
1155            th_obj = self.get_target(th)
1156            errors.extend(th_obj.update_would_cause_errors(th_events))
1157        if len(errors) > 0:
1158            return (False, errors)
1159        # update dependency graph (and undo it if errors)
1160        graph_changes = self.global_dependency_graph.formula_update(
1161            events, include_atoms=False)
1162        if graph_changes:
1163            if self.global_dependency_graph.has_cycle():
1164                # TODO(thinrichs): include path
1165                errors.append(exception.PolicyException(
1166                    "Rules are recursive"))
1167                self.global_dependency_graph.undo_changes(graph_changes)
1168        if len(errors) > 0:
1169            return (False, errors)
1170        # modify execution triggers
1171        self._maintain_triggers()
1172        # figure out relevant triggers
1173        triggers = self.trigger_registry.relevant_triggers(events)
1174        LOG.info("relevant triggers (update): %s",
1175                 ";".join(str(x) for x in triggers))
1176        # signal trigger registry about graph updates
1177        self.trigger_registry.update_dependencies(graph_changes)
1178
1179        # run queries on relevant triggers *before* applying changes
1180        table_triggers = self.trigger_registry.triggers_by_table(triggers)
1181        table_data_old = self._compute_table_contents(table_triggers)
1182        # actually apply the updates
1183        changes = []
1184        for th, th_events in by_theory.items():
1185            changes.extend(self.get_target(th).update(events))
1186        # rerun the trigger queries to check for changes
1187        table_data_new = self._compute_table_contents(table_triggers)
1188        # run triggers if tables changed
1189        for table, triggers in table_triggers.items():
1190            if table_data_old[table] != table_data_new[table]:
1191                for trigger in triggers:
1192                    trigger.callback(table,
1193                                     table_data_old[table],
1194                                     table_data_new[table])
1195        # return non-error and the list of changes
1196        return (True, changes)
1197
1198    def _maintain_triggers(self):
1199        pass
1200
1201    def _actual_events(self, events):
1202        actual = []
1203        for event in events:
1204            th_obj = self.get_target(event.target)
1205            actual.extend(th_obj.actual_events([event]))
1206        return actual
1207
1208    def _compute_table_contents(self, table_policy_pairs):
1209        data = {}   # dict from (table, policy) to set of query results
1210        for table, policy, modal in table_policy_pairs:
1211            th = self.get_target(policy)
1212            queries = self.table_contents_queries(table, policy, modal) or []
1213            data[(table, policy, modal)] = set()
1214            for query in queries:
1215                ans = set(self._select_obj(self.parse1(query), th, False))
1216                data[(table, policy, modal)] |= ans
1217        return data
1218
1219    def _group_events_by_target(self, events):
1220        """Return mapping of targets and events.
1221
1222        Return a dictionary mapping event.target to the list of events
1223        with that target.  Assumes each event.target is a string.
1224        Returns a dictionary from event.target to <list of events>.
1225        """
1226        by_target = {}
1227        for event in events:
1228            if event.target not in by_target:
1229                by_target[event.target] = [event]
1230            else:
1231                by_target[event.target].append(event)
1232        return by_target
1233
1234    def _reroute_events(self, events):
1235        """Events re-routing.
1236
1237        Given list of events with different event.target values,
1238        change each event.target so that the events are routed to the
1239        proper place.
1240        """
1241        by_target = self._group_events_by_target(events)
1242        for target, target_events in by_target.items():
1243            newth = self._compute_route(target_events, target)
1244            for event in target_events:
1245                event.target = newth
1246
1247    def _references_to_policy(self, name):
1248        refs = []
1249        name = name + ":"
1250        for th_obj in self.theory.values():
1251            for rule in th_obj.policy():
1252                if any(table.startswith(name) for table in rule.tablenames()):
1253                    refs.append((name, rule))
1254        return refs
1255
1256    ##########################
1257    # Implementation: queries
1258
1259    # select
1260    def _select_string(self, policy_string, theory, trace):
1261        policy = self.parse(policy_string)
1262        assert (len(policy) == 1), (
1263            "Queries can have only 1 statement: {}".format(
1264                [str(x) for x in policy]))
1265        results = self._select_obj(policy[0], theory, trace)
1266        if trace:
1267            return (compile.formulas_to_string(results[0]), results[1])
1268        else:
1269            return compile.formulas_to_string(results)
1270
1271    def _select_tuple(self, tuple, theory, trace):
1272        return self._select_obj(compile.Literal.create_from_iter(tuple),
1273                                theory, trace)
1274
1275    def _select_obj(self, query, theory, trace):
1276        if trace:
1277            old_tracer = self.get_tracer()
1278            tracer = base.StringTracer()  # still LOG.debugs trace
1279            tracer.trace('*')     # trace everything
1280            self.set_tracer(tracer)
1281            value = set(theory.select(query))
1282            self.set_tracer(old_tracer)
1283            return (value, tracer.get_value())
1284        return set(theory.select(query))
1285
1286    # simulate
1287    def _simulate_string(self, query, theory, sequence, action_theory, delta,
1288                         trace, as_list):
1289        query = self.parse(query)
1290        if len(query) > 1:
1291            raise exception.PolicyException(
1292                "Query %s contained more than 1 rule" % query)
1293        query = query[0]
1294        sequence = self.parse(sequence)
1295        result = self._simulate_obj(query, theory, sequence, action_theory,
1296                                    delta, trace)
1297        if trace:
1298            actual_result = result[0]
1299        else:
1300            actual_result = result
1301        strresult = [str(x) for x in actual_result]
1302        if not as_list:
1303            strresult = " ".join(strresult)
1304        if trace:
1305            return (strresult, result[1])
1306        else:
1307            return strresult
1308
1309    def _simulate_obj(self, query, theory, sequence, action_theory, delta,
1310                      trace):
1311        """Simulate objects.
1312
1313        Both THEORY and ACTION_THEORY are names of theories.
1314        Both QUERY and SEQUENCE are parsed.
1315        """
1316        assert compile.is_datalog(query), "Query must be formula"
1317        # Each action is represented as a rule with the actual action
1318        #    in the head and its supporting data (e.g. options) in the body
1319        assert all(compile.is_extended_datalog(x) for x in sequence), (
1320            "Sequence must be an iterable of Rules")
1321        th_object = self.get_target(theory)
1322
1323        if trace:
1324            old_tracer = self.get_tracer()
1325            tracer = base.StringTracer()  # still LOG.debugs trace
1326            tracer.trace('*')     # trace everything
1327            self.set_tracer(tracer)
1328
1329        # if computing delta, query the current state
1330        if delta:
1331            self.table_log(query.tablename(),
1332                           "** Simulate: Querying %s", query)
1333            oldresult = th_object.select(query)
1334            self.table_log(query.tablename(),
1335                           "Original result of %s is %s",
1336                           query, utility.iterstr(oldresult))
1337
1338        # apply SEQUENCE
1339        self.table_log(query.tablename(), "** Simulate: Applying sequence %s",
1340                       utility.iterstr(sequence))
1341        undo = self.project(sequence, theory, action_theory)
1342
1343        # query the resulting state
1344        self.table_log(query.tablename(), "** Simulate: Querying %s", query)
1345        result = set(th_object.select(query))
1346        self.table_log(query.tablename(), "Result of %s is %s", query,
1347                       utility.iterstr(result))
1348        # rollback the changes
1349        self.table_log(query.tablename(), "** Simulate: Rolling back")
1350        self.project(undo, theory, action_theory)
1351
1352        # if computing the delta, do it
1353        if delta:
1354            result = set(result)
1355            oldresult = set(oldresult)
1356            pos = result - oldresult
1357            neg = oldresult - result
1358            pos = [formula.make_update(is_insert=True) for formula in pos]
1359            neg = [formula.make_update(is_insert=False) for formula in neg]
1360            result = pos + neg
1361        if trace:
1362            self.set_tracer(old_tracer)
1363            return (result, tracer.get_value())
1364        return result
1365
1366    # Helpers
1367
1368    def _react_to_changes(self, changes):
1369        """Filters changes and executes actions contained therein."""
1370        # LOG.debug("react to: %s", iterstr(changes))
1371        actions = self.get_action_names()
1372        formulas = [change.formula for change in changes
1373                    if (isinstance(change, compile.Event)
1374                        and change.is_insert()
1375                        and change.formula.is_atom()
1376                        and change.tablename() in actions)]
1377        # LOG.debug("going to execute: %s", iterstr(formulas))
1378        self.execute(formulas)
1379
1380    def _data_listeners(self):
1381        return [self.theory[self.ENFORCEMENT_THEORY]]
1382
1383    def _compute_route(self, events, theory):
1384        """Compute rerouting.
1385
1386        When a formula is inserted/deleted (in OPERATION) into a THEORY,
1387        it may need to be rerouted to another theory.  This function
1388        computes that rerouting.  Returns a Theory object.
1389        """
1390        self.table_log(None, "Computing route for theory %s and events %s",
1391                       theory.name, utility.iterstr(events))
1392        # Since Enforcement includes Classify and Classify includes Database,
1393        #   any operation on data needs to be funneled into Enforcement.
1394        #   Enforcement pushes it down to the others and then
1395        #   reacts to the results.  That is, we really have one big theory
1396        #   Enforcement + Classify + Database as far as the data is concerned
1397        #   but formulas can be inserted/deleted into each policy individually.
1398        if all([compile.is_atom(event.formula) for event in events]):
1399            if (theory is self.theory[self.CLASSIFY_THEORY] or
1400                    theory is self.theory[self.DATABASE]):
1401                return self.theory[self.ENFORCEMENT_THEORY]
1402        return theory
1403
1404    def project(self, sequence, policy_theory, action_theory):
1405        """Apply the list of updates SEQUENCE.
1406
1407        Apply the list of updates SEQUENCE, where actions are described
1408        in ACTION_THEORY. Return an update sequence that will undo the
1409        projection.
1410
1411        SEQUENCE can include atom insert/deletes, rule insert/deletes,
1412        and action invocations.  Projecting an action only
1413        simulates that action's invocation using the action's description;
1414        the results are therefore only an approximation of executing
1415        actions directly.  Elements of SEQUENCE are just formulas
1416        applied to the given THEORY.  They are NOT Event()s.
1417
1418        SEQUENCE is really a program in a mini-programming
1419        language--enabling results of one action to be passed to another.
1420        Hence, even ignoring actions, this functionality cannot be achieved
1421        by simply inserting/deleting.
1422        """
1423        actth = self.theory[action_theory]
1424        policyth = self.theory[policy_theory]
1425        # apply changes to the state
1426        newth = nonrecursive.NonrecursiveRuleTheory(abbr="Temp")
1427        newth.tracer.trace('*')
1428        actth.includes.append(newth)
1429        # TODO(thinrichs): turn 'includes' into an object that guarantees
1430        #   there are no cycles through inclusion.  Otherwise we get
1431        #   infinite loops
1432        if actth is not policyth:
1433            actth.includes.append(policyth)
1434        actions = self.get_action_names(action_theory)
1435        self.table_log(None, "Actions: %s", utility.iterstr(actions))
1436        undos = []         # a list of updates that will undo SEQUENCE
1437        self.table_log(None, "Project: %s", sequence)
1438        last_results = []
1439        for formula in sequence:
1440            self.table_log(None, "** Updating with %s", formula)
1441            self.table_log(None, "Actions: %s", utility.iterstr(actions))
1442            self.table_log(None, "Last_results: %s",
1443                           utility.iterstr(last_results))
1444            tablename = formula.tablename()
1445            if tablename not in actions:
1446                if not formula.is_update():
1447                    raise exception.PolicyException(
1448                        "Sequence contained non-action, non-update: " +
1449                        str(formula))
1450                updates = [formula]
1451            else:
1452                self.table_log(tablename, "Projecting %s", formula)
1453                # define extension of current Actions theory
1454                if formula.is_atom():
1455                    assert formula.is_ground(), (
1456                        "Projection atomic updates must be ground")
1457                    assert not formula.is_negated(), (
1458                        "Projection atomic updates must be positive")
1459                    newth.define([formula])
1460                else:
1461                    # instantiate action using prior results
1462                    newth.define(last_results)
1463                    self.table_log(tablename, "newth (with prior results) %s",
1464                                   utility.iterstr(newth.content()))
1465                    bindings = actth.top_down_evaluation(
1466                        formula.variables(), formula.body, find_all=False)
1467                    if len(bindings) == 0:
1468                        continue
1469                    grounds = formula.plug_heads(bindings[0])
1470                    grounds = [act for act in grounds if act.is_ground()]
1471                    assert all(not lit.is_negated() for lit in grounds)
1472                    newth.define(grounds)
1473                self.table_log(tablename,
1474                               "newth contents (after action insertion): %s",
1475                               utility.iterstr(newth.content()))
1476                # self.table_log(tablename, "action contents: %s",
1477                #     iterstr(actth.content()))
1478                # self.table_log(tablename, "action.includes[1] contents: %s",
1479                #     iterstr(actth.includes[1].content()))
1480                # self.table_log(tablename, "newth contents: %s",
1481                #     iterstr(newth.content()))
1482                # compute updates caused by action
1483                updates = actth.consequences(compile.is_update)
1484                updates = self.resolve_conflicts(updates)
1485                updates = unify.skolemize(updates)
1486                self.table_log(tablename, "Computed updates: %s",
1487                               utility.iterstr(updates))
1488                # compute results for next time
1489                for update in updates:
1490                    newth.insert(update)
1491                last_results = actth.consequences(compile.is_result)
1492                last_results = set([atom for atom in last_results
1493                                    if atom.is_ground()])
1494            # apply updates
1495            for update in updates:
1496                undo = self.project_updates(update, policy_theory)
1497                if undo is not None:
1498                    undos.append(undo)
1499        undos.reverse()
1500        if actth is not policyth:
1501            actth.includes.remove(policyth)
1502        actth.includes.remove(newth)
1503        return undos
1504
1505    def project_updates(self, delta, theory):
1506        """Project atom/delta rule insertion/deletion.
1507
1508        Takes an atom/rule DELTA with update head table
1509        (i.e. ending in + or -) and inserts/deletes, respectively,
1510        that atom/rule into THEORY after stripping
1511        the +/-. Returns None if DELTA had no effect on the
1512        current state.
1513        """
1514        theory = delta.theory_name() or theory
1515
1516        self.table_log(None, "Applying update %s to %s", delta, theory)
1517        th_obj = self.theory[theory]
1518        insert = delta.tablename().endswith('+')
1519        newdelta = delta.drop_update().drop_theory()
1520        changed = th_obj.update([compile.Event(formula=newdelta,
1521                                               insert=insert)])
1522        if changed:
1523            return delta.invert_update()
1524        else:
1525            return None
1526
1527    def resolve_conflicts(self, atoms):
1528        """If p+(args) and p-(args) are present, removes the p-(args)."""
1529        neg = set()
1530        result = set()
1531        # split atoms into NEG and RESULT
1532        for atom in atoms:
1533            if atom.table.table.endswith('+'):
1534                result.add(atom)
1535            elif atom.table.table.endswith('-'):
1536                neg.add(atom)
1537            else:
1538                result.add(atom)
1539        # add elems from NEG only if their inverted version not in RESULT
1540        for atom in neg:
1541            if atom.invert_update() not in result:  # slow: copying ATOM here
1542                result.add(atom)
1543        return result
1544
1545    def parse(self, string):
1546        return compile.parse(string, theories=self.theory)
1547
1548    def parse1(self, string):
1549        return compile.parse1(string, theories=self.theory)
1550
1551    ##########################
1552    # Helper functions
1553    ##########################
1554
1555    def get_target(self, name):
1556        if name is None:
1557            if len(self.theory) == 1:
1558                name = next(iter(self.theory))
1559            elif len(self.theory) == 0:
1560                raise exception.PolicyException("No policies exist.")
1561            else:
1562                raise exception.PolicyException(
1563                    "Must choose a policy to operate on")
1564        if name not in self.theory:
1565            raise exception.PolicyException("Unknown policy " + str(name))
1566        return self.theory[name]
1567
1568    def _find_policy_name(self, name_or_id):
1569        """Given name or ID, return the name of the policy or KeyError."""
1570        if name_or_id in self.theory:
1571            return name_or_id
1572        for th in self.theory.values():
1573            if th.id == name_or_id:
1574                return th.name
1575        raise KeyError("Policy %s could not be found" % name_or_id)
1576
1577    def get_target_name(self, name):
1578        """Resolve NAME to the name of a proper policy (even if it is None).
1579
1580        Raises PolicyException there is no such policy.
1581        """
1582        return self.get_target(name).name
1583
1584    def get_action_names(self, target):
1585        """Return a list of the names of action tables."""
1586        if target not in self.theory:
1587            return []
1588        actionth = self.theory[target]
1589        actions = actionth.select(self.parse1('action(x)'))
1590        return [action.arguments[0].name for action in actions]
1591
1592    def table_log(self, table, msg, *args):
1593        self.tracer.log(table, "RT    : %s" % msg, *args)
1594
1595    def set_tracer(self, tracer):
1596        if isinstance(tracer, base.Tracer):
1597            self.tracer = tracer
1598            for th in self.theory:
1599                self.theory[th].set_tracer(tracer)
1600        else:
1601            self.tracer = tracer[0]
1602            for th, tracr in tracer[1].items():
1603                if th in self.theory:
1604                    self.theory[th].set_tracer(tracr)
1605
1606    def get_tracer(self):
1607        """Return (Runtime's tracer, dict of tracers for each theory).
1608
1609        Useful so we can temporarily change tracing.
1610        """
1611        d = {}
1612        for th in self.theory:
1613            d[th] = self.theory[th].get_tracer()
1614        return (self.tracer, d)
1615
1616    def debug_mode(self):
1617        tracer = base.Tracer()
1618        tracer.trace('*')
1619        self.set_tracer(tracer)
1620
1621    def production_mode(self):
1622        tracer = base.Tracer()
1623        self.set_tracer(tracer)
1624
1625
1626##############################################################################
1627# ExperimentalRuntime
1628##############################################################################
1629
1630class ExperimentalRuntime (Runtime):
1631    def explain(self, query, tablenames=None, find_all=False, target=None):
1632        """Event handler for explanations.
1633
1634        Given a ground query and a collection of tablenames
1635        that we want the explanation in terms of,
1636        return proof(s) that the query is true. If
1637        FIND_ALL is True, returns list; otherwise, returns single proof.
1638        """
1639        if isinstance(query, six.string_types):
1640            return self.explain_string(
1641                query, tablenames, find_all, self.get_target(target))
1642        elif isinstance(query, tuple):
1643            return self.explain_tuple(
1644                query, tablenames, find_all, self.get_target(target))
1645        else:
1646            return self.explain_obj(
1647                query, tablenames, find_all, self.get_target(target))
1648
1649    def remediate(self, formula):
1650        """Event handler for remediation."""
1651        if isinstance(formula, six.string_types):
1652            return self.remediate_string(formula)
1653        elif isinstance(formula, tuple):
1654            return self.remediate_tuple(formula)
1655        else:
1656            return self.remediate_obj(formula)
1657
1658    def execute(self, action_sequence):
1659        """Event handler for execute:
1660
1661        Execute a sequence of ground actions in the real world.
1662        """
1663        if isinstance(action_sequence, six.string_types):
1664            return self.execute_string(action_sequence)
1665        else:
1666            return self.execute_obj(action_sequence)
1667
1668    def access_control(self, action, support=''):
1669        """Event handler for making access_control request.
1670
1671        ACTION is an atom describing a proposed action instance.
1672        SUPPORT is any data that should be assumed true when posing
1673        the query.  Returns True iff access is granted.
1674        """
1675        # parse
1676        if isinstance(action, six.string_types):
1677            action = self.parse1(action)
1678            assert compile.is_atom(action), "ACTION must be an atom"
1679        if isinstance(support, six.string_types):
1680            support = self.parse(support)
1681        # add support to theory
1682        newth = nonrecursive.NonrecursiveRuleTheory(abbr="Temp")
1683        newth.tracer.trace('*')
1684        for form in support:
1685            newth.insert(form)
1686        acth = self.theory[self.ACCESSCONTROL_THEORY]
1687        acth.includes.append(newth)
1688        # check if action is true in theory
1689        result = len(acth.select(action, find_all=False)) > 0
1690        # allow new theory to be freed
1691        acth.includes.remove(newth)
1692        return result
1693
1694    # explain
1695    def explain_string(self, query_string, tablenames, find_all, theory):
1696        policy = self.parse(query_string)
1697        assert len(policy) == 1, "Queries can have only 1 statement"
1698        results = self.explain_obj(policy[0], tablenames, find_all, theory)
1699        return compile.formulas_to_string(results)
1700
1701    def explain_tuple(self, tuple, tablenames, find_all, theory):
1702        self.explain_obj(compile.Literal.create_from_iter(tuple),
1703                         tablenames, find_all, theory)
1704
1705    def explain_obj(self, query, tablenames, find_all, theory):
1706        return theory.explain(query, tablenames, find_all)
1707
1708    # remediate
1709    def remediate_string(self, policy_string):
1710        policy = self.parse(policy_string)
1711        assert len(policy) == 1, "Queries can have only 1 statement"
1712        return compile.formulas_to_string(self.remediate_obj(policy[0]))
1713
1714    def remediate_tuple(self, tuple, theory):
1715        self.remediate_obj(compile.Literal.create_from_iter(tuple))
1716
1717    def remediate_obj(self, formula):
1718        """Find a collection of action invocations
1719
1720        That if executed result in FORMULA becoming false.
1721        """
1722        actionth = self.theory[self.ACTION_THEORY]
1723        classifyth = self.theory[self.CLASSIFY_THEORY]
1724        # look at FORMULA
1725        if compile.is_atom(formula):
1726            pass  # TODO(tim): clean up unused variable
1727            # output = formula
1728        elif compile.is_regular_rule(formula):
1729            pass  # TODO(tim): clean up unused variable
1730            # output = formula.head
1731        else:
1732            assert False, "Must be a formula"
1733        # grab a single proof of FORMULA in terms of the base tables
1734        base_tables = classifyth.base_tables()
1735        proofs = classifyth.explain(formula, base_tables, False)
1736        if proofs is None:  # FORMULA already false; nothing to be done
1737            return []
1738        # Extract base table literals that make that proof true.
1739        #   For remediation, we assume it suffices to make any of those false.
1740        #   (Leaves of proof may not be literals or may not be written in
1741        #    terms of base tables, despite us asking for base tables--
1742        #    because of negation.)
1743        leaves = [leaf for leaf in proofs[0].leaves()
1744                  if (compile.is_atom(leaf) and
1745                      leaf.table in base_tables)]
1746        self.table_log(None, "Leaves: %s", utility.iterstr(leaves))
1747        # Query action theory for abductions of negated base tables
1748        actions = self.get_action_names()
1749        results = []
1750        for lit in leaves:
1751            goal = lit.make_positive()
1752            if lit.is_negated():
1753                goal.table = goal.table + "+"
1754            else:
1755                goal.table = goal.table + "-"
1756            # return is a list of goal :- act1, act2, ...
1757            # This is more informative than query :- act1, act2, ...
1758            for abduction in actionth.abduce(goal, actions, False):
1759                results.append(abduction)
1760        return results
1761
1762    ##########################
1763    # Execute actions
1764
1765    def execute_string(self, actions_string):
1766        self.execute_obj(self.parse(actions_string))
1767
1768    def execute_obj(self, actions):
1769        """Executes the list of ACTION instances one at a time.
1770
1771        For now, our execution is just logging.
1772        """
1773        LOG.debug("Executing: %s", utility.iterstr(actions))
1774        assert all(compile.is_atom(action) and action.is_ground()
1775                   for action in actions)
1776        action_names = self.get_action_names()
1777        assert all(action.table in action_names for action in actions)
1778        for action in actions:
1779            if not action.is_ground():
1780                if self.logger is not None:
1781                    self.logger.warn("Unground action to execute: %s", action)
1782                continue
1783            if self.logger is not None:
1784                self.logger.info("%s", action)
1785
1786##############################################################################
1787# Engine that operates on the DSE
1788##############################################################################
1789
1790
1791class PolicySubData (object):
1792    def __init__(self, trigger):
1793        self.table_trigger = trigger
1794        self.to_add = ()
1795        self.to_rem = ()
1796        self.dataindex = trigger.policy + ":" + trigger.tablename
1797
1798    def trigger(self):
1799        return self.table_trigger
1800
1801    def changes(self):
1802        result = []
1803        for row in self.to_add:
1804            event = compile.Event(formula=row, insert=True)
1805            result.append(event)
1806        for row in self.to_rem:
1807            event = compile.Event(formula=row, insert=False)
1808            result.append(event)
1809        return result
1810
1811
1812class DseRuntime (Runtime, data_service.DataService):
1813    def __init__(self, name):
1814        Runtime.__init__(self)
1815        data_service.DataService.__init__(self, name)
1816        self.name = name
1817        self.msg = None
1818        self.last_policy_change = None
1819        self.policySubData = {}
1820        self.log_actions_only = cfg.CONF.enable_execute_action
1821        self.add_rpc_endpoint(DseRuntimeEndpoints(self))
1822        self.periodic_tasks = None
1823        self.sync_thread = None
1824
1825    def extend_schema(self, service_name, schema):
1826        newschema = {}
1827        for key, value in schema:
1828            newschema[service_name + ":" + key] = value
1829        super(DseRuntime, self).extend_schema(self, newschema)
1830
1831    def receive_policy_update(self, msg):
1832        LOG.debug("received policy-update msg %s",
1833                  utility.iterstr(msg.body.data))
1834        # update the policy and subscriptions to data tables.
1835        self.last_policy_change = self.process_policy_update(msg.body.data)
1836
1837    def process_policy_update(self, events, persistent=False):
1838        LOG.debug("process_policy_update %s" % events)
1839        # body_only so that we don't subscribe to tables in the head
1840        oldtables = self.tablenames(body_only=True)
1841        result = Runtime.process_policy_update(self, events,
1842                                               persistent=persistent)
1843        newtables = self.tablenames(body_only=True)
1844        self.update_table_subscriptions(oldtables, newtables)
1845        return result
1846
1847    def initialize_table_subscriptions(self):
1848        """Initialize table subscription.
1849
1850        Once policies have all been loaded, this function subscribes to
1851        all the necessary tables.  See UPDATE_TABLE_SUBSCRIPTIONS as well.
1852        """
1853        self.update_table_subscriptions(set(), self.tablenames())
1854
1855    def update_table_subscriptions(self, oldtables, newtables):
1856        """Update table subscription.
1857
1858        Change the subscriptions from OLDTABLES to NEWTABLES, ensuring
1859        to load all the appropriate services.
1860        """
1861        add = newtables - oldtables
1862        rem = oldtables - newtables
1863        LOG.debug("Tables:: Old: %s, new: %s, add: %s, rem: %s",
1864                  oldtables, newtables, add, rem)
1865        # subscribe to the new tables (loading services as required)
1866        for table in add:
1867            if not self.reserved_tablename(table):
1868                (service, tablename) = compile.Tablename.parse_service_table(
1869                    table)
1870                if service is not None:
1871                    LOG.debug("Subscribing to new (service, table): (%s, %s)",
1872                              service, tablename)
1873                    self.subscribe(service, tablename)
1874
1875        # unsubscribe from the old tables
1876        for table in rem:
1877            (service, tablename) = compile.Tablename.parse_service_table(table)
1878            if service is not None:
1879                LOG.debug("Unsubscribing to new (service, table): (%s, %s)",
1880                          service, tablename)
1881                self.unsubscribe(service, tablename)
1882
1883    # Note(thread-safety): blocking function
1884    def execute_action(self, service_name, action, action_args):
1885        """Event handler for action execution.
1886
1887        :param service_name: openstack service to perform the action on,
1888        e.g. 'nova', 'neutron'
1889        :param action: action to perform on service, e.g. an API call
1890        :param action_args: positional-args and named-args in format:
1891            {'positional': ['p_arg1', 'p_arg2'],
1892            'named': {'name1': 'n_arg1', 'name2': 'n_arg2'}}.
1893        """
1894        if not self.log_actions_only:
1895            LOG.info("action %s is called with args %s on %s, but "
1896                     "current configuration doesn't allow Congress to "
1897                     "execute any action.", action, action_args, service_name)
1898            return
1899
1900        # Log the execution
1901        LOG.info("%s:: executing: %s:%s on %s",
1902                 self.name, service_name, action, action_args)
1903        if self.logger is not None:
1904            pos_args = ''
1905            if 'positional' in action_args:
1906                pos_args = ", ".join(str(x) for x in action_args['positional'])
1907            named_args = ''
1908            if 'named' in action_args:
1909                named_args = ", ".join(
1910                    "%s=%s" % (key, val)
1911                    for key, val in action_args['named'].items())
1912            delimit = ''
1913            if pos_args and named_args:
1914                delimit = ', '
1915            self.logger.info(
1916                "Executing %s:%s(%s%s%s)",
1917                service_name, action, pos_args, delimit, named_args)
1918
1919        # execute the action on a service in the DSE
1920        if not self.service_exists(service_name):
1921            raise exception.PolicyException(
1922                "Service %s not found" % service_name)
1923        if not action:
1924            raise exception.PolicyException("Action not found")
1925        LOG.info("Sending request(%s:%s), args = %s",
1926                 service_name, action, action_args)
1927        # Note(thread-safety): blocking call
1928        self._rpc(service_name, action, args=action_args)
1929
1930    def pub_policy_result(self, table, olddata, newdata):
1931        """Callback for policy table triggers."""
1932        LOG.debug("grabbing policySubData[%s]", table)
1933        policySubData = self.policySubData[table]
1934        policySubData.to_add = newdata - olddata
1935        policySubData.to_rem = olddata - newdata
1936        LOG.debug("Table Data:: Old: %s, new: %s, add: %s, rem: %s",
1937                  olddata, newdata, policySubData.to_add, policySubData.to_rem)
1938
1939        # TODO(dse2): checks needed that all literals are facts
1940        # TODO(dse2): should we support modals and other non-fact literals?
1941        # convert literals to rows for dse2
1942        newdata = [lit.argument_names() for lit in newdata]
1943        self.publish(policySubData.dataindex, newdata)
1944
1945    def get_snapshot(self, table_name):
1946        # print("agnostic policy engine get_snapshot(%s); %s" % (
1947        #     table_name, self.policySubData[table]))
1948        (policy, tablename) = compile.Tablename.parse_service_table(table_name)
1949        data = self.get_row_data(tablename, policy, trace=False)
1950        data = [record['data'] for record in data]
1951        return data
1952
1953    def prepush_processor(self, data, dataindex, type=None):
1954        """Called before push.
1955
1956        Takes as input the DATA that the receiver needs and returns
1957        the payload for the message. If this is a regular publication
1958        message, make the payload just the delta; otherwise, make the
1959        payload the entire table.
1960        """
1961        # This routine basically ignores DATA and sends a delta
1962        # of policy table (i.e. dataindex) changes part of the state.
1963        LOG.debug("prepush_processor: dataindex <%s> data: %s", dataindex,
1964                  data)
1965        # if not a regular publication, just return the original data
1966        if type != 'pub':
1967            LOG.debug("prepush_processor: returned original data")
1968            if type == 'sub' and data is None:
1969                # Always want to send initialization of []
1970                return []
1971            return data
1972        # grab deltas to publish to subscribers
1973        (policy, tablename) = compile.Tablename.parse_service_table(dataindex)
1974        result = self.policySubData[(tablename, policy, None)].changes()
1975        if len(result) == 0:
1976            # Policy engine expects an empty update to be an init msg
1977            # So if delta is empty, return None, which signals
1978            # the message should not be sent.
1979            result = None
1980            text = "None"
1981        else:
1982            text = utility.iterstr(result)
1983        LOG.debug("prepush_processor for <%s> returning with %s items",
1984                  dataindex, text)
1985        return result
1986
1987    def _maintain_triggers(self):
1988        # ensure there is a trigger registered to execute actions
1989        curr_tables = set(self.global_dependency_graph.tables_with_modal(
1990            'execute'))
1991        # add new triggers
1992        for table in curr_tables:
1993            LOG.debug("%s:: checking for missing trigger table %s",
1994                      self.name, table)
1995            if table not in self.execution_triggers:
1996                (policy, tablename) = compile.Tablename.parse_service_table(
1997                    table)
1998                LOG.debug("creating new trigger for policy=%s, table=%s",
1999                          policy, tablename)
2000                trig = self.trigger_registry.register_table(
2001                    tablename, policy,
2002                    lambda table, old, new: self._execute_table(
2003                        policy, tablename, old, new),
2004                    modal='execute')
2005                self.execution_triggers[table] = trig
2006        # remove triggers no longer needed
2007        #    Using copy of execution_trigger keys so we can delete inside loop
2008        for table in self.execution_triggers.copy().keys():
2009            LOG.debug("%s:: checking for stale trigger table %s",
2010                      self.name, table)
2011            if table not in curr_tables:
2012                LOG.debug("removing trigger for table %s", table)
2013                try:
2014                    self.trigger_registry.unregister(
2015                        self.execution_triggers[table])
2016                    del self.execution_triggers[table]
2017                except KeyError:
2018                    LOG.exception(
2019                        "Tried to unregister non-existent trigger: %s", table)
2020
2021    # Note(thread-safety): blocking function
2022    def _execute_table(self, theory, table, old, new):
2023        # LOG.info("execute_table(theory=%s, table=%s, old=%s, new=%s",
2024        #          theory, table, ";".join(str(x) for x in old),
2025        #          ";".join(str(x) for x in new))
2026        service, tablename = compile.Tablename.parse_service_table(table)
2027        service = service or theory
2028        for newlit in new - old:
2029            args = [term.name for term in newlit.arguments]
2030            LOG.info("%s:: on service %s executing %s on %s",
2031                     self.name, service, tablename, args)
2032            try:
2033                # Note(thread-safety): blocking call
2034                self.execute_action(service, tablename, {'positional': args})
2035            except exception.PolicyException as e:
2036                LOG.error(str(e))
2037
2038    def start_policy_synchronizer(self):
2039        callables = [(self.synchronize, None, {})]
2040        self.periodic_tasks = periodics.PeriodicWorker(callables)
2041        if self._running:
2042            self.sync_thread = eventlet.spawn_n(self.periodic_tasks.start)
2043
2044    def stop_policy_synchronizer(self):
2045        # stop the periodic task worker
2046        if self.periodic_tasks:
2047            self.periodic_tasks.stop()
2048            self.periodic_tasks.wait()
2049            self.periodic_tasks = None
2050        # kill synchronizer greenthread
2051        if self.sync_thread:
2052            eventlet.greenthread.kill(self.sync_thread)
2053            self.sync_thread = None
2054
2055    def stop(self):
2056        self.stop_policy_synchronizer()
2057        super(DseRuntime, self).stop()
2058
2059    @periodics.periodic(spacing=cfg.CONF.datasource_sync_period,
2060                        run_immediately=True)
2061    def synchronize(self):
2062        try:
2063            LOG.info("Synchronizing policies on node %s", self.node.node_id)
2064            self.synchronize_policies()
2065        except Exception:
2066            LOG.exception("synchronize_policies failed")
2067        try:
2068            self.synchronize_rules()
2069        except Exception:
2070            LOG.exception('synchronize_rules failed')
2071
2072    # eventually we should remove the action theory as a default,
2073    #   but we need to update the docs and tutorials
2074    def create_default_policies(self):
2075        # sync policies first before multiple pe, tries to create same policies
2076        self.synchronize_policies()
2077        if self.DEFAULT_THEORY not in self.theory:
2078            self.persistent_create_policy(name=self.DEFAULT_THEORY,
2079                                          desc='default policy')
2080
2081        if self.ACTION_THEORY not in self.theory:
2082            self.persistent_create_policy(name=self.ACTION_THEORY,
2083                                          kind=base.ACTION_POLICY_TYPE,
2084                                          desc='default action policy')
2085
2086    # Note(thread-safety): blocking function
2087    def _rpc(self, service_name, action, args):
2088        """Overloading the DseRuntime version of _rpc so it uses dse2."""
2089        # TODO(ramineni): This is called only during execute_action, added
2090        # the same function name for compatibility with old arch
2091        args = {'action': action, 'action_args': args}
2092
2093        def execute_once():
2094            return self.rpc(service_name, 'request_execute', args,
2095                            timeout=cfg.CONF.dse.long_timeout, retry=0)
2096
2097        def execute_retry():
2098            timeout = cfg.CONF.dse.execute_action_retry_timeout
2099            start_time = time.time()
2100            end_time = start_time + timeout
2101            while timeout <= 0 or time.time() < end_time:
2102                try:
2103                    return self.rpc(
2104                        service_name, 'request_execute', args,
2105                        timeout=cfg.CONF.dse.long_timeout, retry=0)
2106                except (messaging_exceptions.MessagingTimeout,
2107                        messaging_exceptions.MessageDeliveryFailure):
2108                    LOG.warning('DSE failure executing action %s with '
2109                                'arguments %s. Retrying.',
2110                                action, args['action_args'])
2111            LOG.error('Failed to executing action %s with arguments %s',
2112                      action, args['action_args'])
2113
2114        # long timeout for action execution because actions can take a while
2115        if not cfg.CONF.dse.execute_action_retry:
2116            # Note(thread-safety): blocking call
2117            #   Only when thread pool at capacity
2118            eventlet.spawn_n(execute_once)
2119            eventlet.sleep(0)
2120        else:
2121            # Note(thread-safety): blocking call
2122            #   Only when thread pool at capacity
2123            eventlet.spawn_n(execute_retry)
2124            eventlet.sleep(0)
2125
2126    def service_exists(self, service_name):
2127        return self.is_valid_service(service_name)
2128
2129    def receive_data(self, publisher, table, data, is_snapshot=False):
2130        """Event handler for when a dataservice publishes data.
2131
2132        That data can either be the full table (as a list of tuples)
2133        or a delta (a list of Events).
2134        """
2135        LOG.debug("received data msg for %s:%s", publisher, table)
2136        if not is_snapshot:
2137            to_add = data[0]
2138            to_del = data[1]
2139            result = []
2140            for row in to_del:
2141                formula = compile.Literal.create_from_table_tuple(table, row)
2142                event = compile.Event(formula=formula, insert=False)
2143                result.append(event)
2144            for row in to_add:
2145                formula = compile.Literal.create_from_table_tuple(table, row)
2146                event = compile.Event(formula=formula, insert=True)
2147                result.append(event)
2148            self.receive_data_update(publisher, table, result)
2149            return
2150
2151        # if empty data, assume it is an init msg, since noop otherwise
2152        if len(data) == 0:
2153            self.receive_data_full(publisher, table, data)
2154        else:
2155            # grab an item from any iterable
2156            dataelem = next(iter(data))
2157            if isinstance(dataelem, compile.Event):
2158                self.receive_data_update(publisher, table, data)
2159            else:
2160                self.receive_data_full(publisher, table, data)
2161
2162    def receive_data_full(self, publisher, table, data):
2163        """Handler for when dataservice publishes full table."""
2164        LOG.debug("received full data msg for %s:%s. %s",
2165                  publisher, table, utility.iterstr(data))
2166        # Use a generator to avoid instantiating all these Facts at once.
2167        facts = (compile.Fact(table, row) for row in data)
2168        self.initialize_tables([table], facts, target=publisher)
2169
2170    def receive_data_update(self, publisher, table, data):
2171        """Handler for when dataservice publishes a delta."""
2172        LOG.debug("received update data msg for %s:%s: %s",
2173                  publisher, table, utility.iterstr(data))
2174        events = data
2175        for event in events:
2176            assert compile.is_atom(event.formula), (
2177                "receive_data_update received non-atom: " +
2178                str(event.formula))
2179            # prefix tablename with data source
2180            event.target = publisher
2181        (permitted, changes) = self.update(events)
2182        if not permitted:
2183            raise exception.CongressException(
2184                "Update not permitted." + '\n'.join(str(x) for x in changes))
2185        else:
2186            LOG.debug("update data msg for %s from %s caused %d "
2187                      "changes: %s", table, publisher, len(changes),
2188                      utility.iterstr(changes))
2189            if table in self.theory[publisher].tablenames():
2190                rows = self.theory[publisher].content([table])
2191                LOG.debug("current table: %s", utility.iterstr(rows))
2192
2193    def on_first_subs(self, tables):
2194        """handler for policy table subscription
2195
2196        when a previously non-subscribed table gains a subscriber, register a
2197        trigger for the tables and publish table results when there is
2198        updates.
2199        """
2200        for table in tables:
2201            (policy, tablename) = compile.Tablename.parse_service_table(
2202                table)
2203            # we only care about policy table subscription
2204            if policy is None:
2205                return
2206
2207            if not (tablename, policy, None) in self.policySubData:
2208                trig = self.trigger_registry.register_table(
2209                    tablename,
2210                    policy,
2211                    self.pub_policy_result)
2212                self.policySubData[
2213                    (tablename, policy, None)] = PolicySubData(trig)
2214
2215    def on_no_subs(self, tables):
2216        """Remove triggers when tables have no subscribers."""
2217        for table in tables:
2218            (policy, tablename) = compile.Tablename.parse_service_table(table)
2219            if (tablename, policy, None) in self.policySubData:
2220                # release resource if no one cares about it any more
2221                sub = self.policySubData.pop((tablename, policy, None))
2222                self.trigger_registry.unregister(sub.trigger())
2223        return True
2224
2225    def set_schema(self, name, schema, complete=False):
2226        old_tables = self.tablenames(body_only=True)
2227        super(DseRuntime, self).set_schema(name, schema, complete)
2228        new_tables = self.tablenames(body_only=True)
2229        self.update_table_subscriptions(old_tables, new_tables)
2230
2231    def synchronize_rules(self):
2232        LOG.info("Synchronizing rules on node %s", self.node.node_id)
2233
2234        # Read rules from DB.
2235        configured_rules = [{'rule': r.rule,
2236                             'id': r.id,
2237                             'comment': r.comment,
2238                             'name': r.name,
2239                             'policy_name': r.policy_name}
2240                            for r in db_policy_rules.get_policy_rules()]
2241
2242        # Read rules from engine
2243        policies = {n: self.policy_object(n) for n in self.policy_names()}
2244        active_policy_rules = []
2245        for policy_name, policy in policies.items():
2246            if policy.kind != base.DATASOURCE_POLICY_TYPE:
2247                for active_rule in policy.content():
2248                    active_policy_rules.append(
2249                        {'rule': active_rule.original_str,
2250                         'id': active_rule.id,
2251                         'comment': active_rule.comment,
2252                         'name': active_rule.name,
2253                         'policy_name': policy_name})
2254
2255        # ALEX: the Rule object does not have fields like the rule-string or
2256        # id or comment.  We can add those fields to the Rule object, as long
2257        # as we don't add them to the Fact because there are many fact
2258        # instances.  If a user tries to create a lot of Rules, they are
2259        # probably doing something wrong and should use a datasource driver
2260        # instead.
2261
2262        changes = []
2263        for r in configured_rules:
2264            if r not in active_policy_rules:
2265                LOG.debug("adding rule %s", str(r))
2266                parsed_rule = self.parse1(r['rule'])
2267                parsed_rule.set_id(r['id'])
2268                parsed_rule.set_name(r['name'])
2269                parsed_rule.set_comment(r['comment'])
2270                parsed_rule.set_original_str(r['rule'])
2271
2272                event = compile.Event(formula=parsed_rule,
2273                                      insert=True,
2274                                      target=r['policy_name'])
2275                changes.append(event)
2276
2277        for r in active_policy_rules:
2278            if r not in configured_rules:
2279                # Note(ekcs): temporary work-around to avoid failure due to a
2280                # mysterious compile or agnostic problem inserting None rule
2281                # into engine memory
2282                if r['rule'] is None:
2283                    continue
2284                LOG.debug("removing rule %s", str(r))
2285                parsed_rule = self.parse1(r['rule'])
2286                parsed_rule.set_id(r['id'])
2287                parsed_rule.set_name(r['name'])
2288                parsed_rule.set_comment(r['comment'])
2289                parsed_rule.set_original_str(r['rule'])
2290
2291                event = compile.Event(formula=parsed_rule,
2292                                      insert=False,
2293                                      target=r['policy_name'])
2294                changes.append(event)
2295        permitted, changes = self.process_policy_update(changes)
2296        LOG.debug("synchronize_rules, permitted %d, made %d changes on "
2297                  "node %s", permitted, len(changes), self.node.node_id)
2298
2299
2300class DseRuntimeEndpoints(object):
2301    """RPC endpoints exposed by DseRuntime."""
2302
2303    def __init__(self, dse):
2304        self.dse = dse
2305
2306    # Note(thread-safety): blocking function
2307    def persistent_create_policy(self, context, name=None, id_=None,
2308                                 abbr=None, kind=None, desc=None):
2309        # Note(thread-safety): blocking call
2310        return self.dse.persistent_create_policy(name, id_, abbr, kind, desc)
2311
2312    # Note(thread-safety): blocking function
2313    def persistent_delete_policy(self, context, name_or_id):
2314        # Note(thread-safety): blocking call
2315        return self.dse.persistent_delete_policy(name_or_id)
2316
2317    # Note(thread-safety): blocking function
2318    def persistent_get_policies(self, context):
2319        # Note(thread-safety): blocking call
2320        return self.dse.persistent_get_policies()
2321
2322    # Note(thread-safety): blocking function
2323    def persistent_get_policy(self, context, id_):
2324        # Note(thread-safety): blocking call
2325        return self.dse.persistent_get_policy(id_)
2326
2327    # Note(thread-safety): blocking function
2328    def persistent_get_rule(self, context, id_, policy_name):
2329        # Note(thread-safety): blocking call
2330        return self.dse.persistent_get_rule(id_, policy_name)
2331
2332    # Note(thread-safety): blocking function
2333    def persistent_get_rules(self, context, policy_name):
2334        # Note(thread-safety): blocking call
2335        return self.dse.persistent_get_rules(policy_name)
2336
2337    # Note(thread-safety): blocking function
2338    def persistent_insert_rule(self, context, policy_name, str_rule, rule_name,
2339                               comment):
2340        # Note(thread-safety): blocking call
2341        return self.dse.persistent_insert_rule(
2342            policy_name, str_rule, rule_name, comment)
2343
2344    # Note(thread-safety): blocking function
2345    def persistent_delete_rule(self, context, id_, policy_name_or_id):
2346        # Note(thread-safety): blocking call
2347        return self.dse.persistent_delete_rule(id_, policy_name_or_id)
2348
2349    # Note(thread-safety): blocking function
2350    def persistent_load_policies(self, context):
2351        # Note(thread-safety): blocking call
2352        return self.dse.persistent_load_policies()
2353
2354    def simulate(self, context, query, theory, sequence, action_theory,
2355                 delta=False, trace=False, as_list=False):
2356        return self.dse.simulate(query, theory, sequence, action_theory,
2357                                 delta, trace, as_list)
2358
2359    def get_tablename(self, context, source_id, table_id):
2360        return self.dse.get_tablename(source_id, table_id)
2361
2362    def get_tablenames(self, context, source_id):
2363        return self.dse.get_tablenames(source_id)
2364
2365    def get_status(self, context, source_id, params):
2366        return self.dse.get_status(source_id, params)
2367
2368    def get_row_data(self, context, table_id, source_id, trace=False):
2369        return self.dse.get_row_data(table_id, source_id, trace)
2370
2371    # Note(thread-safety): blocking function
2372    def execute_action(self, context, service_name, action, action_args):
2373        # Note(thread-safety): blocking call
2374        return self.dse.execute_action(service_name, action, action_args)
2375
2376    def delete_policy(self, context, name, disallow_dangling_refs=False):
2377        return self.dse.delete_policy(name, disallow_dangling_refs)
2378
2379    def initialize_datasource(self, context, name, schema):
2380        return self.dse.initialize_datasource(name, schema)