path: root/onap-dcae-dcaepolicy-lib/onap_dcae_dcaepolicy_lib/dcae_policy.py

# org.onap.dcae
# ================================================================================
# Copyright (c) 2017-2018 AT&T Intellectual Property. All rights reserved.
# ================================================================================
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# ============LICENSE_END=========================================================
#
# ECOMP is a trademark and service mark of AT&T Intellectual Property.

"""dcae_policy contains decorators for the policy lifecycle in cloudify"""

import json
from copy import deepcopy
from functools import wraps
import traceback

from cloudify import ctx
from cloudify.context import NODE_INSTANCE
from cloudify.exceptions import NonRecoverableError

from .utils import Utils

POLICIES = 'policies'
POLICY_FILTERS = 'policy_filters'
POLICIES_FILTERED = 'policies_filtered'
POLICY_APPLY_ORDER = 'policy_apply_order'
POLICY_APPLY_ORDER_CLAUSE = 'policy_apply_order_clause'
POLICY_DEFAULTED_FIELDS = 'policy_defaulted_fields'

POLICY_ID = 'policy_id'
POLICY_BODY = 'policy_body'
POLICY_VERSION = "policyVersion"
POLICY_CONFIG = 'config'
POLICY_CONFIG_CONTENT = 'content'
APPLICATION_CONFIG = 'application_config'
POLICY_FILTER = 'policy_filter'
POLICY_FILTER_ID = 'policy_filter_id'
POLICY_TYPE = 'policyType'
POLICY_TYPE_MICROSERVICE = 'MicroService'
POLICY_CONFIG_MS = "Config_MS"

POLICY_PERSISTENT = 'policy_persistent'
DCAE_POLICY_TYPE = 'dcae.nodes.policy'
DCAE_POLICIES_TYPE = 'dcae.nodes.policies'
POLICY_MESSAGE_TYPE = 'policy'

class Policies(object):
    """static class for policy operations"""
    _updated_policies = {}
    _removed_policies = {}

    @staticmethod
    def _init():
        """init static members"""
        Policies._updated_policies = {}
        Policies._removed_policies = {}

    @staticmethod
    def _get_config_from_policy(policy):
        """returns the config field from policy object"""
        policy_body = (policy or {}).get(POLICY_BODY)
        if not policy_body:
            return

        policy_config = policy_body.get(POLICY_CONFIG)
        policy_type = policy_body.get(POLICY_TYPE)
        if not policy_type:
            policy_id = policy.get(POLICY_ID)
            if policy_id:
                policy_type_pos = policy_id.rfind(".") + 1
                if policy_type_pos and policy_id.startswith(POLICY_CONFIG_MS, policy_type_pos):
                    policy_type = POLICY_TYPE_MICROSERVICE

        if policy_type == POLICY_TYPE_MICROSERVICE and isinstance(policy_config, dict):
            return policy_config.get(POLICY_CONFIG_CONTENT)

        return policy_config

    @staticmethod
    def _store_policy_apply_order_clause(policy_apply_order_path):
        """
        Find the field :policy_apply_order_path: that is provided as an optional parameter
        to gather_policies_to_node decorator.

        Parse the field-pathes found in the :policy_apply_order_path: field.

        Store the result into :runtime_properties[POLICY_APPLY_ORDER_CLAUSE]:

        Example:
            policy_apply_order_path = "docker_config:policy:apply_order"
            will take the list of field-pathes from the field
                properties:
                    docker_config:
                        policy:
                            apply_order:
        """
        if not policy_apply_order_path:
            return

        policy_apply_order_clause = Utils.get_field_value(
            dict(ctx.node.properties),
            policy_apply_order_path
        )

        if not policy_apply_order_clause:
            ctx.logger.warn("not found policy_apply_order_path: {0} in node.properties"
                            .format(policy_apply_order_path))
            return

        ctx.instance.runtime_properties[POLICY_APPLY_ORDER_CLAUSE] = policy_apply_order_clause

    @staticmethod
    def _set_policy_apply_order(policies):
        """
        Calculates, sorts and stores the policy_apply_order for policies.

        Sorting is done based on the list of field-pathes
        specified in :runtime_properties[POLICY_APPLY_ORDER_CLAUSE]:

        The apply_order field is expected to be formatted as the list of strings.
        Each string can contain the path to the field inside the policy_body object
        with the same delimiter of ":" (semicolon).
        To convert the field to decimal, use ::number suffix after the field name.
        To specify the descending order, add "desc" after the whitespace.

        Example:
            apply_order = ["matchingConditions:priority::number desc",
                           "config:foo desc nulls-last", "config:db_client"]

        this will return the policies starting with the highest decimal value in "priority"
        field inside the "matchingConditions".
        Then the policies with the same "priority" values will be sorted in descending order
        by the string value in the field "foo" found inside the "config".
        Then the policies with the same "priority" and "foo" values will be sorted
        by the string value in the field "db_client" found inside the "config".
        Then the policies with the same "priority" and "foo" and "db_client" field values will
        further be always sorted by "policy_id" - no need to specify that in apply_order.
        Sorting by policy_id insures the uniqueness and predictability of the policy apply_order.

        An invalid field-path will result in the value of None that brings this record upfront.
        """
        policy_apply_order = [policy_id
                              for (policy_id, policy) in policies.iteritems()
                              if Policies._get_config_from_policy(policy)]

        if not policy_apply_order:
            ctx.instance.runtime_properties[POLICY_APPLY_ORDER] = policy_apply_order
            return

        policy_apply_order.sort()

        policy_apply_order_clause = ctx.instance.runtime_properties.get(POLICY_APPLY_ORDER_CLAUSE)
        if not policy_apply_order_clause:
            ctx.instance.runtime_properties[POLICY_APPLY_ORDER] = policy_apply_order
            return

        if not isinstance(policy_apply_order_clause, list):
            policy_apply_order_clause = [policy_apply_order_clause]

        for clause_item in reversed(policy_apply_order_clause):
            f_path, f_type, desc, n_last = Utils.parse_clause_item(clause_item)

            if not f_path:
                continue

            policy_apply_order.sort(
                key=lambda policy_id, fpath=f_path, ftype=f_type, reverse=desc, nulls_last=n_last:
                Utils.key_with_none_in_sort(
                    reverse, nulls_last,
                    Utils.get_field_value(
                        policies.get(policy_id, {}).get(POLICY_BODY, {}),
                        fpath,
                        field_type=ftype
                    )
                ), reverse=desc
            )

        ctx.instance.runtime_properties[POLICY_APPLY_ORDER] = policy_apply_order

    @staticmethod
    def _set_policy_defaulted_fields(policies):
        """
        Keeps track and stores the dict of field names of removed policies into
        :runtime_properties[POLICY_DEFAULTED_FIELDS]:
        """
        policy_defaulted_fields = ctx.instance.runtime_properties.get(POLICY_DEFAULTED_FIELDS, {})
        policy_defaulted_fields.update(
            (k, True)
            for policy in Policies._removed_policies.itervalues()
            for k in Policies._get_config_from_policy(policy) or {}
        )
        if policies:
            for policy in policies.itervalues():
                for k in Policies._get_config_from_policy(policy) or {}:
                    if k in policy_defaulted_fields:
                        del policy_defaulted_fields[k]

        ctx.instance.runtime_properties[POLICY_DEFAULTED_FIELDS] = policy_defaulted_fields

    @staticmethod
    def _set_policies(policies):
        """
        store the :policies: in :runtime_properties[POLICIES]:

        and build an index on policies into :runtime_properties[POLICY_APPLY_ORDER]:

        and keep track of fields from previously :removed: policy-configs in
            :runtime_properties[POLICY_DEFAULTED_FIELDS]: to reset them to default values
            on merging the policies into config
        """
        Policies._set_policy_defaulted_fields(policies)
        if not policies:
            if POLICIES in ctx.instance.runtime_properties:
                del ctx.instance.runtime_properties[POLICIES]
            if POLICY_APPLY_ORDER in ctx.instance.runtime_properties:
                del ctx.instance.runtime_properties[POLICY_APPLY_ORDER]
            return

        ctx.instance.runtime_properties[POLICIES] = policies
        Policies._set_policy_apply_order(policies)

    @staticmethod
    def gather_policies_to_node(policy_apply_order_path=None):
        """
        decorate with @Policies.gather_policies_to_node to
        gather the policies from dcae.nodes.policy nodes this node depends on.

        Places the policies into runtime_properties["policies"].

        Call Policies.calc_latest_application_config() to apply policies onto application_config.
        """
        def gather_policies_decorator(func):
            """the decorator"""
            if not func:
                return

            def add_policy(policy_id, policy, policy_persistent, policies):
                """only add the latest version of policy to policies"""
                prev_policy = policies.get(policy_id)
                prev_policy_persistent = (prev_policy or {}).get(POLICY_PERSISTENT)

                if not prev_policy \
                or (policy_persistent and not prev_policy_persistent) \
                or (policy_persistent == prev_policy_persistent and policy.get(POLICY_BODY)):
                    policy = deepcopy(policy)
                    policy[POLICY_PERSISTENT] = policy_persistent
                    policies[policy_id] = policy

            def gather_policy(target, policies):
                """adds the policy from dcae.nodes.policy node to policies"""
                if DCAE_POLICY_TYPE not in target.node.type_hierarchy:
                    return
                policy_id = target.node.properties.get(POLICY_ID)
                if not policy_id:
                    return True
                policy = deepcopy(dict(target.node.properties))
                policy_body = target.instance.runtime_properties.get(POLICY_BODY)
                if policy_body:
                    policy[POLICY_BODY] = policy_body

                add_policy(policy_id, policy, True, policies)
                return True

            def gather_policies(target, policies, policy_filters):
                """adds the policies and policy-filter from dcae.nodes.policies node to policies"""
                if DCAE_POLICIES_TYPE not in target.node.type_hierarchy:
                    return

                property_policy_filter = target.node.properties.get(POLICY_FILTER)
                if property_policy_filter:
                    policy_filter = dict(
                        (k, v) for (k, v) in dict(property_policy_filter).iteritems()
                        if v or isinstance(v, (int, float))
                    )
                    if policy_filter:
                        policy_filters[target.instance.id] = {
                            POLICY_FILTER_ID : target.instance.id,
                            POLICY_FILTER : deepcopy(policy_filter)
                        }

                filtered_policies = target.instance.runtime_properties.get(POLICIES_FILTERED)
                if not filtered_policies or not isinstance(filtered_policies, dict):
                    return True
                for (policy_id, policy) in filtered_policies.iteritems():
                    add_policy(policy_id, policy, False, policies)
                return True

            @wraps(func)
            def wrapper(*args, **kwargs):
                """gather and save the policies from dcae.nodes.policy nodes this node related to"""
                if ctx.type != NODE_INSTANCE:
                    raise NonRecoverableError("can only invoke gather_policies_to_node on node")

                try:
                    Policies._store_policy_apply_order_clause(policy_apply_order_path)

                    policies = {}
                    policy_filters = {}
                    for rel in ctx.instance.relationships:
                        _ = gather_policy(rel.target, policies) \
                         or gather_policies(rel.target, policies, policy_filters)

                    Policies._set_policies(policies)
                    if policy_filters:
                        ctx.instance.runtime_properties[POLICY_FILTERS] = policy_filters
                except Exception as ex:
                    error = "Failed to set the policies {0}".format(str(ex))
                    ctx.logger.error("{0}: {1}".format(error, traceback.format_exc()))
                    raise NonRecoverableError(error)

                return func(*args, **kwargs)
            return wrapper
        return gather_policies_decorator

    @staticmethod
    def _update_policies(updated_policies, added_policies, removed_policies):
        """
        filter and update policies in runtime_properties
        and return the ordered filtered list of changed_policies
        """
        Policies._init()

        if not updated_policies and not removed_policies and not added_policies:
            ctx.logger.error(
                "update_policies_on_ctx - no updated, added, or removed policies received")
            return

        updated_policies = updated_policies or []
        added_policies = added_policies or {}
        removed_policies = removed_policies or []

        ctx.logger.info("updated_policies: {0}, added_policies: {1}, removed_policies: {2}"
                        .format(json.dumps(updated_policies),
                                json.dumps(added_policies),
                                json.dumps(removed_policies)))

        policies = ctx.instance.runtime_properties.get(POLICIES, {})
        policy_filters = ctx.instance.runtime_properties.get(POLICY_FILTERS, {})

        for policy_id in removed_policies:
            removed_policy = policies.get(policy_id)
            if removed_policy:
                Policies._removed_policies[policy_id] = deepcopy(removed_policy)
                if not removed_policy.get(POLICY_PERSISTENT):
                    del policies[policy_id]
                elif POLICY_BODY in removed_policy:
                    del policies[policy_id][POLICY_BODY]

        new_policies = dict((p_id, policy)
                            for policy_filter_id in policy_filters
                            for (p_id, policy) in added_policies
                            .get(policy_filter_id, {})
                            .get(POLICIES, {})
                            .iteritems())

        ctx.logger.info("new_policies: {0}".format(json.dumps(new_policies)))

        for (policy_id, policy) in new_policies.iteritems():
            deployed_policy = policies.get(policy_id)
            if not deployed_policy:
                policies[policy_id] = policy
                Policies._updated_policies[policy_id] = policy
                continue
            updated_policies.append(policy)

        skipped = {"ignored" : [], "unexpected" : [], "same" : []}
        for policy in updated_policies:
            policy_id = policy.get(POLICY_ID)
            if not policy_id or policy_id not in policies:
                skipped["ignored"].append(policy)
                continue

            updated_policy_body = policy.get(POLICY_BODY, {})
            updated_policy_version = updated_policy_body.get(POLICY_VERSION)
            if not updated_policy_version or POLICY_CONFIG not in updated_policy_body:
                skipped["unexpected"].append(policy)
                continue

            deployed_policy = policies.get(policy_id)
            deployed_policy_version = deployed_policy.get(POLICY_BODY, {}).get(POLICY_VERSION)
            if updated_policy_version == deployed_policy_version:
                skipped["same"].append(policy)
                continue

            policies[policy_id][POLICY_BODY] = updated_policy_body
            Policies._updated_policies[policy_id] = policy

        if skipped["same"] or skipped["ignored"] or skipped["unexpected"]:
            ctx.logger.info("skipped policies: {0}".format(json.dumps(skipped)))

        if Policies._updated_policies or Policies._removed_policies:
            Policies._set_policies(policies)

    @staticmethod
    def update_policies_on_node(configs_only=True):
        """decorate each policy_update operation with @Policies.update_policies_on_node to
        filter out the updated_policies to only what applies to the current node instance,
        update runtime_properties["policies"]

        :configs_only: - set to True if expect to see only the config in updated_policies
            instead of the whole policy object (False)

        Passes through the filtered list of updated_policies that apply to the current node instance

        :updated_policies: contains the list of changed policy-configs when configs_only=True.
        """
        def update_policies_decorator(func):
            """actual decorator"""
            if not func:
                return

            @wraps(func)
            def wrapper(updated_policies=None,
                        added_policies=None,
                        removed_policies=None,
                        **kwargs):
                """update matching policies on the node"""
                if ctx.type != NODE_INSTANCE:
                    raise NonRecoverableError("can only invoke update_policies_on_node on node")

                try:
                    Policies._update_policies(updated_policies, added_policies, removed_policies)

                    updated_policies = deepcopy(Policies._get_ordered_policies(
                        selected_policies=Policies._updated_policies
                    ))
                    removed_policies = deepcopy(Policies._removed_policies.values())

                    if configs_only:
                        updated_policies = Utils.remove_empties(
                            [Policies._get_config_from_policy(policy)
                             for policy in updated_policies or []]
                        )
                        removed_policies = [Policies._get_config_from_policy(policy)
                                            for policy in removed_policies or []]

                except Exception as ex:
                    error = "Failed to update the policies {0}".format(str(ex))
                    ctx.logger.error("{0}: {1}".format(error, traceback.format_exc()))
                    raise NonRecoverableError(error)

                if updated_policies or removed_policies:
                    return func(updated_policies, removed_policies=removed_policies, **kwargs)
            return wrapper
        return update_policies_decorator

    @staticmethod
    def _get_ordered_policies(selected_policies=None):
        """
        returns the ordered list of selected policies from the runtime policies
        ordered by policy_id values
        """
        policies = ctx.instance.runtime_properties.get(POLICIES)
        apply_order = ctx.instance.runtime_properties.get(POLICY_APPLY_ORDER)
        if not policies or not apply_order:
            return []

        if selected_policies is None:
            return [policies[policy_id] for policy_id in apply_order]

        if not selected_policies:
            return []

        return [policies[policy_id] for policy_id in apply_order if policy_id in selected_policies]

    @staticmethod
    def get_policy_configs():
        """
        returns the list of policy configs from the runtime policies
        ordered by policy_id values
        """
        if ctx.type != NODE_INSTANCE:
            return []

        ordered_policies = Policies._get_ordered_policies()
        return Utils.remove_empties(
            [Policies._get_config_from_policy(policy)
             for policy in ordered_policies]
        )

    @staticmethod
    def shallow_merge_policies_into(config, default_config=None):
        """
        shallow merge the :policy configs: (dict) into :config: that is expected to be a dict.

        the fields listed in :runtime_properties[POLICY_DEFAULTED_FIELDS]:
        that where ever changed by policy-configs are initially reset to default values
        found in :default_config: or :node.properties[APPLICATION_CONFIG]:
            on merging the policies into config
        """
        if config is None:
            config = {}

        policy_configs = Policies.get_policy_configs()

        if not config or not isinstance(config, dict):
            ctx.logger.warn("unexpected config {0} to merge the policy_configs {1} into" \
                .format(json.dumps(config), json.dumps(policy_configs or [])))
            return config

        defaulted_fields = ctx.instance.runtime_properties.get(POLICY_DEFAULTED_FIELDS)
        if defaulted_fields:
            if default_config is None or not isinstance(default_config, dict):
                default_config = dict(ctx.node.properties.get(APPLICATION_CONFIG, {}))
                ctx.logger.info("using default_config from node.properties[{0}]: {1}"
                                .format(APPLICATION_CONFIG, json.dumps(default_config)))
            if default_config and isinstance(default_config, dict):
                for defaulted_field in defaulted_fields:
                    if defaulted_field in default_config and defaulted_field in config:
                        config[defaulted_field] = deepcopy(default_config.get(defaulted_field))

                ctx.logger.info("inited config {0} on {1} {2} from {3}"
                                .format(json.dumps(config),
                                        POLICY_DEFAULTED_FIELDS,
                                        json.dumps(defaulted_fields),
                                        json.dumps(default_config)))

        if not policy_configs:
            ctx.logger.warn("no policies to merge to config {0}".format(json.dumps(config)))
            return config

        for policy_config in policy_configs:
            if not isinstance(policy_config, dict):
                ctx.logger.warn("skipped unexpected format of policy_config {0} for config: {1}" \
                    .format(json.dumps(policy_config), json.dumps(config)))
                continue

            ctx.logger.info("applying policy_config {0} to config {1}" \
                .format(json.dumps(policy_config), json.dumps(config)))
            for (policy_key, policy_value) in policy_config.iteritems():
                if policy_key not in config or policy_value is None:
                    ctx.logger.warn("skipped unexpected policy({0}, {1}) for config: {2}" \
                        .format(policy_key, json.dumps(policy_value), json.dumps(config)))
                    continue
                config[policy_key] = deepcopy(policy_value)

        return config

    @staticmethod
    def calc_latest_application_config(application_config_name=None):
        """
        shallow merge the policy configs (dict) into config that is expected to be a dict

        if :application_config_name: is not provided,
        the runtime property :application_config: on the node instance is used as initial config
        """
        if not application_config_name:
            application_config_name = APPLICATION_CONFIG

        config = deepcopy(dict(ctx.instance.runtime_properties.get(application_config_name, {})))
        if not config:
            config = deepcopy(dict(ctx.node.properties.get(application_config_name, {})))

        ctx.logger.info("going to merge policies over {0}: {1}" \
            .format(application_config_name, json.dumps(config)))

        return Policies.shallow_merge_policies_into(config)