voltha/core/config/config_node.py

#
# Copyright 2016 the original author or authors.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
from collections import OrderedDict
from copy import copy

from jsonpatch import JsonPatch
from jsonpatch import make_patch

from common.utils.json_format import MessageToDict
from voltha.core.config.config_branch import ConfigBranch
from voltha.core.config.config_proxy import CallbackType, ConfigProxy
from voltha.core.config.config_rev import is_proto_message, children_fields, \
    ConfigRevision, access_rights
from voltha.core.config.config_rev_persisted import PersistedConfigRevision
from voltha.protos import third_party
from voltha.protos import meta_pb2


class MergeConflictException(Exception):
    pass


def message_to_dict(m):
    return MessageToDict(m, True, True, False)


def check_access_violation(new_msg, old_msg):
    """Raise ValueError if attempt is made to change a read-only field"""
    access_map = access_rights(new_msg.__class__)
    violated_fields = []
    for field_name, access in access_map.iteritems():
        if access == meta_pb2.READ_ONLY:
            if getattr(new_msg, field_name) != getattr(old_msg, field_name):
                violated_fields.append(field_name)
    if violated_fields:
        raise ValueError('Cannot change read-only field(s) %s' %
                         ', '.join('"%s"' % f for f in violated_fields))


class ConfigNode(object):
    """
    Represents a configuration node which can hold a number of revisions
    of the configuration for this node.
    When the configuration changes, the new version is appended to the
    node.
    Initial data must be a protobuf message and it will determine the type of
    this node.
    """
    __slots__ = (
        '_root',  # ref to root node
        '_type',  # node type, as __class__ of protobuf message
        '_branches',  # dict of transaction branches and a default (committed)
                      # branch
        '_tags',  # dict of tag-name to ref of ConfigRevision
        '_proxy',  # ref to proxy observer or None if no proxy assigned
        '_auto_prune'
    )

    def __init__(self, root, initial_data, auto_prune=True, txid=None):
        self._root = root
        self._branches = {}
        self._tags = {}
        self._proxy = None
        self._auto_prune = auto_prune

        if isinstance(initial_data, type):
            self._type = initial_data
        elif is_proto_message(initial_data):
            self._type = initial_data.__class__
            self._initialize(copy(initial_data), txid)
        else:
            raise NotImplementedError()

    def _mknode(self, *args, **kw):
        return ConfigNode(self._root, *args, **kw)

    def _mkrev(self, *args, **kw):
        return self._root.mkrev(*args, **kw)

    def _initialize(self, data, txid):
        # separate external children data away from locally stored data
        # based on child_node annotations in protobuf
        children = {}
        for field_name, field in children_fields(self._type).iteritems():
            field_value = getattr(data, field_name)
            if field.is_container:
                if field.key:
                    children[field_name] = od = OrderedDict()
                    for v in field_value:
                        rev = self._mknode(v, txid=txid).latest
                        key = getattr(v, field.key)
                        if key in od:
                            raise ValueError('Duplicate key "{}"'.format(key))
                        od[key] = rev
                else:
                    children[field_name] = [
                        self._mknode(v, txid=txid).latest for v in field_value]
            else:
                children[field_name] = [
                    self._mknode(field_value, txid=txid).latest]
            data.ClearField(field_name)

        branch = ConfigBranch(self, auto_prune=self._auto_prune)
        rev = self._mkrev(branch, data, children)
        self._make_latest(branch, rev)
        self._branches[txid] = branch

    # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ accessors ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
    # these convenience short-cuts only work for the committed branch

    @property
    def revisions(self):
        return [r._hash for r in self._branches[None]._revs.itervalues()]

    @property
    def latest(self):
        return self._branches[None]._latest

    def __getitem__(self, hash):
        return self._branches[None]._revs[hash]

    # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~ get operation ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

    def get(self, path=None, hash=None, depth=0, deep=False, txid=None):

        # depth preparation
        if deep:
            depth = -1

        # path preparation
        path = '' if path is None else path
        while path.startswith('/'):
            path = path[1:]

        # determine branch; if lookup fails, it is ok to use default branch
        branch = self._branches.get(txid, None) or self._branches[None]

        # determine rev
        if hash is not None:
            rev = branch._revs[hash]
        else:
            rev = branch.latest

        return self._get(rev, path, depth)

    def _get(self, rev, path, depth):

        if not path:
            return self._do_get(rev, depth)

        # ... otherwise
        name, _, path = path.partition('/')
        field = children_fields(self._type)[name]
        if field.is_container:
            if field.key:
                children_od = rev._children[name]
                if path:
                    # need to escalate further
                    key, _, path = path.partition('/')
                    child_rev = children_od[field.key_from_str(key)]
                    child_node = child_rev.node
                    return child_node._get(child_rev, path, depth)
                else:
                    # we are the node of interest
                    response = []
                    for child_rev in children_od.itervalues():
                        child_node = child_rev.node
                        value = child_node._do_get(child_rev, depth)
                        response.append(value)
                    return response
            else:
                if path:
                    raise LookupError(
                        'Cannot index into container with no key defined')
                response = []
                for child_rev in rev._children[name]:
                    child_node = child_rev.node
                    value = child_node._do_get(child_rev, depth)
                    response.append(value)
                return response
        else:
            child_rev = rev._children[name][0]
            child_node = child_rev.node
            return child_node._get(child_rev, path, depth)

    def _do_get(self, rev, depth):
        msg = rev.get(depth)
        if self._proxy is not None:
            msg = self._proxy.invoke_callbacks(CallbackType.GET, msg)
        return msg

    # ~~~~~~~~~~~~~~~~~~~~~~~~~~~ update operation ~~~~~~~~~~~~~~~~~~~~~~~~~~~~

    def update(self, path, data, strict=False, txid=None, mk_branch=None):

        while path.startswith('/'):
            path = path[1:]

        try:
            branch = self._branches[txid]
        except KeyError:
            branch = mk_branch(self)

        if not path:
            return self._do_update(branch, data, strict)

        rev = branch._latest  # change is always made to the latest
        name, _, path = path.partition('/')
        field = children_fields(self._type)[name]
        if field.is_container:
            if not path:
                raise ValueError('Cannot update a list')
            if field.key:
                key, _, path = path.partition('/')
                key = field.key_from_str(key)
                children_od = copy(rev._children[name])
                child_rev = children_od[key]
                child_node = child_rev.node
                new_child_rev = child_node.update(
                    path, data, strict, txid, mk_branch)
                if new_child_rev.hash == child_rev.hash:
                    # no change, we can return
                    return branch._latest
                if getattr(new_child_rev.data, field.key) != key:
                    raise ValueError('Cannot change key field')
                children_od[key] = new_child_rev
                rev = rev.update_children(name, children_od, branch)
                self._make_latest(branch, rev)
                return rev
            else:
                raise ValueError('Cannot index into container with no keys')

        else:
            child_rev = rev._children[name][0]
            child_node = child_rev.node
            new_child_rev = child_node.update(
                path, data, strict, txid, mk_branch)
            rev = rev.update_children(name, [new_child_rev], branch)
            self._make_latest(branch, rev)
            return rev

    def _do_update(self, branch, data, strict):
        if not isinstance(data, self._type):
            raise ValueError(
                '"{}" is not a valid data type for this node'.format(
                    data.__class__.__name__))
        self._test_no_children(data)
        if self._proxy is not None:
            self._proxy.invoke_callbacks(CallbackType.PRE_UPDATE, data)

        if branch._latest.data != data:
            if strict:
                # check if attempt is made to change read-only field
                check_access_violation(data, branch._latest.data)
            rev = branch._latest.update_data(data, branch)
            self._make_latest(branch, rev,
                              ((CallbackType.POST_UPDATE, rev.data),))
            return rev
        else:
            return branch._latest

    def _make_latest(self, branch, rev, change_announcements=()):
        branch._latest = rev
        if rev.hash not in branch._revs:
            branch._revs[rev.hash] = rev

        # announce only if this is main branch
        if change_announcements and branch._txid is None and \
                        self._proxy is not None:
            for change_type, data in change_announcements:
                self._proxy.invoke_callbacks(
                    change_type, data, proceed_on_errors=1)

    # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ add operation ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

    def add(self, path, data, txid=None, mk_branch=None):
        while path.startswith('/'):
            path = path[1:]
        if not path:
            raise ValueError('Cannot add to non-container node')

        try:
            branch = self._branches[txid]
        except KeyError:
            branch = mk_branch(self)

        rev = branch._latest  # change is always made to latest
        name, _, path = path.partition('/')
        field = children_fields(self._type)[name]
        if field.is_container:
            if not path:
                # we do need to add a new child to the field
                if field.key:
                    if self._proxy is not None:
                        self._proxy.invoke_callbacks(
                            CallbackType.PRE_ADD, data)
                    children_od = copy(rev._children[name])
                    key = getattr(data, field.key)
                    if key in children_od:
                        raise ValueError('Duplicate key "{}"'.format(key))
                    child_rev = self._mknode(data).latest
                    children_od[key] = child_rev
                    rev = rev.update_children(name, children_od, branch)
                    self._make_latest(branch, rev,
                                      ((CallbackType.POST_ADD, data),))
                    return rev
                else:
                    # adding to non-keyed containers not implemented yet
                    raise ValueError('Cannot add to non-keyed container')
            else:
                if field.key:
                    # need to escalate
                    key, _, path = path.partition('/')
                    key = field.key_from_str(key)
                    children_od = copy(rev._children[name])
                    child_rev = children_od[key]
                    child_node = child_rev.node
                    new_child_rev = child_node.add(path, data, txid, mk_branch)
                    children_od[key] = new_child_rev
                    rev = rev.update_children(name, children_od, branch)
                    self._make_latest(branch, rev)
                    return rev
                else:
                    raise ValueError(
                        'Cannot index into container with no keys')
        else:
            raise ValueError('Cannot add to non-container field')

    # ~~~~~~~~~~~~~~~~~~~~~~~~~~~ remove operation ~~~~~~~~~~~~~~~~~~~~~~~~~~~~

    def remove(self, path, txid=None, mk_branch=None):
        while path.startswith('/'):
            path = path[1:]
        if not path:
            raise ValueError('Cannot remove from non-container node')

        try:
            branch = self._branches[txid]
        except KeyError:
            branch = mk_branch(self)

        rev = branch._latest  # change is always made to latest
        name, _, path = path.partition('/')
        field = children_fields(self._type)[name]
        if field.is_container:
            if not path:
                raise ValueError("Cannot remove without a key")
            if field.key:
                key, _, path = path.partition('/')
                key = field.key_from_str(key)
                if path:
                    # need to escalate
                    children_od = copy(rev._children[name])
                    child_rev = children_od[key]
                    child_node = child_rev.node
                    new_child_rev = child_node.remove(path, txid, mk_branch)
                    children_od[key] = new_child_rev
                    rev = rev.update_children(name, children_od, branch)
                    self._make_latest(branch, rev)
                    return rev
                else:
                    # need to remove from this very node
                    children_od = copy(rev._children[name])
                    if self._proxy is not None:
                        data = children_od[field.key_from_str(key)].data
                        self._proxy.invoke_callbacks(
                            CallbackType.PRE_REMOVE, data)
                        post_anno = ((CallbackType.POST_REMOVE, data),)
                    else:
                        post_anno = ()
                    del children_od[field.key_from_str(key)]
                    rev = rev.update_children(name, children_od, branch)
                    self._make_latest(branch, rev, post_anno)
                    return rev
            else:
                raise ValueError('Cannot remove from non-keyed container')
        else:
            raise ValueError('Cannot remove non-conatiner field')

    # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ Branching ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

    def _mk_txbranch(self, txid):
        branch_point = self._branches[None].latest
        branch = ConfigBranch(self, txid, branch_point)
        self._branches[txid] = branch
        return branch

    def _del_txbranch(self, txid):
        del self._branches[txid]

    # def can_txbranch_be_merged(self, txid):
    #     try:
    #         self._merge_txbranch(txid, dry_run=True)
    #     except MergeConflictException:
    #         return False
    #     else:
    #         return True

    def _merge_txbranch(self, txid, dry_run=False):
        """
        Make latest in branch to be latest in the common branch, but only
        if no conflict is detected. Conflict is where the txbranch branch
        point no longer matches the latest in the default branch. This has
        to be verified recursively.
        """

        """
        A transaction branch can be merged only if none of the following
        happened with the master branch since the fork rev:
        - the local data was changed both in the incoming node and in the
          default branch since the branch point, and they differ now
        - both branches changed the same children nodes in any way (local or
          deep)
        """

        announcements = []

        def _get_od_changes(lst1, lst2):
            assert isinstance(lst2, dict)
            added_keys = [k for k in lst2.iterkeys() if k not in lst1]
            removed_keys = [k for k in lst1.iterkeys() if k not in lst2]
            changed_keys = [k for k in lst1.iterkeys()
                            if k in lst2 and lst1[k].hash != lst2[k].hash]
            return added_keys, removed_keys, changed_keys

        def _get_changes(lst1, lst2):
            if isinstance(lst1, dict):
                return _get_od_changes(lst1, lst2)
            assert isinstance(lst1, list)
            assert isinstance(lst2, list)
            set1 = set(lst1)
            set2 = set(lst2)
            added = set2.difference(set1)
            removed = set1.difference(set2)
            changed = set()  # no such thing in plain (unkeyed) lists
            return added, removed, changed

        def _escalate(child_rev):
            child_branch = child_rev._branch
            if child_branch._txid == txid:
                child_rev = child_branch._node._merge_txbranch(txid, dry_run)
            return child_rev

        def _escalate_list(src_list):
            if isinstance(src_list, list):
                lst = []
                for child_rev in src_list:
                    lst.append(_escalate(child_rev))
                return lst
            else:  # OrderedDict
                od = OrderedDict()
                for key, child_rev in src_list.iteritems():
                    od[key] = _escalate(child_rev)
                return od

        def _add(dst, rev_or_key, src):
            if isinstance(dst, list):
                dst.append(_escalate(rev_or_key))
                announcements.append((CallbackType.POST_ADD, rev_or_key.data))
            else:  # OrderedDict key, data is in lst
                rev = src[rev_or_key]
                dst[rev_or_key] = _escalate(rev)
                announcements.append((CallbackType.POST_ADD, rev.data))

        def _remove(dst, rev_or_key):
            if isinstance(dst, list):
                dst.remove(rev_or_key)
                announcements.append((CallbackType.POST_REMOVE, rev_or_key))
            else:
                rev = dst[rev_or_key]
                del dst[rev_or_key]
                announcements.append((CallbackType.POST_REMOVE, rev.data))

        src_branch = self._branches[txid]
        dst_branch = self._branches[None]

        fork_rev = src_branch.origin  # rev from which src branch was made
        src_rev = src_branch.latest  # head rev of source branch
        dst_rev = dst_branch.latest  # head rev of target branch

        # deal with config data first
        if dst_rev._config is fork_rev._config:
            # no change in master, accept src if different
            config_changed = dst_rev._config != src_rev._config
        else:
            if dst_rev._config.hash != src_rev._config.hash:
                raise MergeConflictException('Config collision')
            config_changed = True

        new_children = copy(dst_rev._children)
        for field_name, field in children_fields(self._type).iteritems():
            fork_list = fork_rev._children[field_name]
            src_list = src_rev._children[field_name]
            dst_list = dst_rev._children[field_name]
            if 0: #dst_list == fork_list:
                # no change in master, accept src if different
                if src_list != fork_list:
                    new_children[field_name] = _escalate_list(src_list)
            else:
                src_added, src_removed, src_changed = _get_changes(
                    fork_list, src_list)
                dst_added, dst_removed, dst_changed = _get_changes(
                    fork_list, dst_list)

                lst = copy(new_children[field_name])
                for to_add in src_added:
                    # we cannot add if it has been added and is different
                    if to_add in dst_added:
                        # this can happen only to keyed containers
                        assert isinstance(src_list, dict)
                        if src_list[to_add].hash != dst_list[to_add].hash:
                            raise MergeConflictException(
                                'Cannot add because it has been added and '
                                'different'
                            )
                    _add(lst, to_add, src_list)
                for to_remove in src_removed:
                    # we cannot remove if it has changed in dst
                    if to_remove in dst_changed:
                        raise MergeConflictException(
                            'Cannot remove because it has changed')
                    if to_remove not in dst_removed:
                        _remove(lst, to_remove)
                for to_change in src_changed:
                    # we cannot change if it was removed in dst
                    if to_change in dst_removed:
                        raise MergeConflictException(
                            'Cannot change because it has been removed')
                    # change can only be in keyed containers (OrderedDict)
                    lst[to_change] = _escalate(src_list[to_change])
                new_children[field_name] = lst

        if not dry_run:
            rev = src_rev if config_changed else dst_rev
            rev = rev.update_all_children(new_children, dst_branch)
            if config_changed:
                announcements.append((CallbackType.POST_UPDATE, rev.data))
            self._make_latest(dst_branch, rev, announcements)
            del self._branches[txid]
            return rev

    # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ Diff utility ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

    def diff(self, hash1, hash2=None, txid=None):
        branch = self._branches[txid]
        rev1 = branch[hash1]
        rev2 = branch[hash2] if hash2 else branch._latest
        if rev1.hash == rev2.hash:
            return JsonPatch([])
        else:
            dict1 = message_to_dict(rev1.data)
            dict2 = message_to_dict(rev2.data)
            return make_patch(dict1, dict2)

    # ~~~~~~~~~~~~~~~~~~~~~~~~~~~ Tagging utility ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

    def tag(self, tag, hash=None):
        branch = self._branches[None]  # tag only what has been committed
        rev = branch._latest if hash is None else branch._revs[hash]
        self._tags[tag] = rev
        return self

    @property
    def tags(self):
        return sorted(self._tags.iterkeys())

    def by_tag(self, tag):
        """
        Return revision based on tag
        :param tag: previously registered tag value
        :return: revision object
        """
        return self._tags[tag]

    def diff_by_tag(self, tag1, tag2):
        return self.diff(self._tags[tag1].hash, self._tags[tag2].hash)

    def delete_tag(self, tag):
        del self._tags[tag]

    def delete_tags(self, *tags):
        for tag in tags:
            del self._tags[tag]

    def prune_untagged(self):
        branch = self._branches[None]
        keep = set(rev.hash for rev in self._tags.itervalues())
        keep.add(branch._latest.hash)
        for hash in branch._revs.keys():
            if hash not in keep:
                del branch._revs[hash]
        return self

    # ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ Internals ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

    def _test_no_children(self, data):
        for field_name, field in children_fields(self._type).items():
            field_value = getattr(data, field_name)
            if field.is_container:
                if len(field_value):
                    raise NotImplementedError(
                        'Cannot update external children')
            else:
                if data.HasField(field_name):
                    raise NotImplementedError(
                        'Cannot update externel children')

    # ~~~~~~~~~~~~~~~~~~~~~~~~~~~ Node proxy ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

    def get_proxy(self, path, exclusive=False):
        return self._get_proxy(path, self, path, exclusive)

    def _get_proxy(self, path, root, full_path, exclusive):
        while path.startswith('/'):
            path = path[1:]
        if not path:
            return self._mk_proxy(root, full_path, exclusive)

        # need to escalate
        rev = self._branches[None]._latest
        name, _, path = path.partition('/')
        field = children_fields(self._type)[name]
        if field.is_container:
            if not path:
                raise ValueError('Cannot proxy a container field')
            if field.key:
                key, _, path = path.partition('/')
                children_od = rev._children[name]
                child_rev = children_od[key]
                child_node = child_rev.node
                return child_node._get_proxy(path, root, full_path, exclusive)

            raise ValueError('Cannot index into container with no keys')

        else:
            child_rev = rev._children[name][0]
            child_node = child_rev.node
            return child_node._get_proxy(path, root, full_path, exclusive)

    def _mk_proxy(self, root, full_path, exclusive):
        if self._proxy is None:
            self._proxy = ConfigProxy(root, self, full_path, exclusive)
        else:
            if self._proxy.exclusive:
                raise ValueError('Node is already owned exclusively')
        return self._proxy

    # ~~~~~~~~~~~~~~~~~~~~~~~~ Persistence loading ~~~~~~~~~~~~~~~~~~~~~~~~~~~~

    def load_latest(self, latest_hash):

        root = self._root
        kv_store = root._kv_store

        branch = ConfigBranch(self, self._auto_prune)
        rev = PersistedConfigRevision.load(
            branch, kv_store, self._type, latest_hash)
        self._make_latest(branch, rev)
        self._branches[None] = branch