diff options
Diffstat (limited to 'onap-dcae-dcaepolicy-lib/onap_dcae_dcaepolicy_lib/dcae_policy.py')
| -rw-r--r-- | onap-dcae-dcaepolicy-lib/onap_dcae_dcaepolicy_lib/dcae_policy.py | 461 |
1 files changed, 89 insertions, 372 deletions
diff --git a/onap-dcae-dcaepolicy-lib/onap_dcae_dcaepolicy_lib/dcae_policy.py b/onap-dcae-dcaepolicy-lib/onap_dcae_dcaepolicy_lib/dcae_policy.py index b064157..5a3a0b2 100644 --- a/onap-dcae-dcaepolicy-lib/onap_dcae_dcaepolicy_lib/dcae_policy.py +++ b/onap-dcae-dcaepolicy-lib/onap_dcae_dcaepolicy_lib/dcae_policy.py @@ -1,4 +1,3 @@ -# org.onap.dcae # ================================================================================ # Copyright (c) 2017-2018 AT&T Intellectual Property. All rights reserved. # ================================================================================ @@ -29,26 +28,18 @@ from cloudify.context import NODE_INSTANCE from cloudify.exceptions import NonRecoverableError from .policies_output import PoliciesOutput -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' @@ -69,168 +60,68 @@ class Policies(object): 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: + def _set_policies(policies): + """store the :policies: in :runtime_properties[POLICIES]:""" + if not policies: + if POLICIES in ctx.instance.runtime_properties: + del ctx.instance.runtime_properties[POLICIES] 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 + ctx.instance.runtime_properties[POLICIES] = policies @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 + 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 @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 + 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 - 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 + Policies._add_policy(policy_id, policy, True, policies) + return True @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( - (field_name, True) - for policy in Policies._removed_policies.itervalues() - for field_name in Policies._get_config_from_policy(policy) or {} - ) - if policies: - for policy in policies.itervalues(): - for field_name in Policies._get_config_from_policy(policy) or {}: - if field_name in policy_defaulted_fields: - del policy_defaulted_fields[field_name] - - 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] + 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 - ctx.instance.runtime_properties[POLICIES] = policies - Policies._set_policy_apply_order(policies) + 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(): + Policies._add_policy(policy_id, policy, False, policies) + return True @staticmethod def _get_policy_bodies_dict(policies): @@ -243,7 +134,7 @@ class Policies(object): ) @staticmethod - def gather_policies_to_node(policy_apply_order_path=None): + def gather_policies_to_node(): """ decorate with @Policies.gather_policies_to_node() to gather the policies from dcae.nodes.policy nodes this node depends on. @@ -251,65 +142,12 @@ class Policies(object): Places the policies into runtime_properties["policies"]. Stores <scn>:policies data in consul-kv - - 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""" @@ -319,13 +157,11 @@ class Policies(object): policies_outputted = False policy_bodies = [] 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._gather_policy(rel.target, policies) \ + or Policies._gather_policies(rel.target, policies, policy_filters) Policies._set_policies(policies) if policy_filters: @@ -357,8 +193,7 @@ class Policies(object): 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") + ctx.logger.error("update_policies_on_ctx - no updated, added, or removed policies received") return updated_policies = updated_policies or [] @@ -432,20 +267,21 @@ class Policies(object): PoliciesOutput.store_policies(ACTION_UPDATED, policy_bodies) @staticmethod - def update_policies_on_node(configs_only=False): + def update_policies_on_node(): """ - decorate each policy_update operation with @Policies.update_policies_on_node to + 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"] updates <scn>:policies data in consul-kv - :configs_only: - set to True if expect to see only the config in updated_policies - instead of the whole policy_body 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. + :updated_policies: contains the list of changed policy_bodies + + :removed_policies: contains the list of removed policy_bodies + + :policies: contains the list of current policy_bodies """ def update_policies_decorator(func): """actual decorator""" @@ -453,10 +289,7 @@ class Policies(object): return @wraps(func) - def wrapper(updated_policies=None, - added_policies=None, - removed_policies=None, - **kwargs): + 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") @@ -464,24 +297,13 @@ class Policies(object): try: Policies._update_policies(updated_policies, added_policies, removed_policies) - updated_policies = deepcopy(Policies._get_ordered_policies( + updated_policies = Policies.get_policy_bodies( 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] - ) - removed_policies = [Policies._get_config_from_policy(policy) - for policy in removed_policies] - else: - updated_policies = Utils.remove_empties( - [policy.get(POLICY_BODY) for policy in updated_policies] - ) - removed_policies = [policy.get(POLICY_BODY) - for policy in removed_policies] + ) + + removed_policies = Policies.get_policy_bodies( + selected_policies=Policies._removed_policies + ) except Exception as ex: error = "Failed to update the policies {0}".format(str(ex)) @@ -489,18 +311,17 @@ class Policies(object): raise NonRecoverableError(error) if updated_policies or removed_policies: - return func( - updated_policies, - removed_policies=removed_policies, - policies=Policies.get_policy_bodies(), - **kwargs) + return func(updated_policies, + removed_policies=removed_policies, + policies=Policies.get_policy_bodies(), + **kwargs) return wrapper return update_policies_decorator @staticmethod def cleanup_policies_on_node(func): """ - decorate each policy_update operation with @Policies.cleanup_policies_on_node to + decorate each delete operation with @Policies.cleanup_policies_on_node to remove <scn>:policies data in consul-kv """ if not func: @@ -508,128 +329,24 @@ class Policies(object): @wraps(func) def wrapper(**kwargs): - """cleanup policies in consul-kv""" - if ctx.type != NODE_INSTANCE: - raise NonRecoverableError("can only invoke cleanup_policies_on_node on node") - - try: - PoliciesOutput.delete_policies() - except Exception as ex: - error = "Failed to cleanup policies in consul-kv {0}".format(str(ex)) - ctx.logger.error("{0}: {1}".format(error, traceback.format_exc())) + """remove policies in consul-kv""" + if ctx.type == NODE_INSTANCE: + try: + PoliciesOutput.delete_policies() + except Exception as ex: + error = "Failed to cleanup policies in consul-kv {0}".format(str(ex)) + ctx.logger.error("{0}: {1}".format(error, traceback.format_exc())) return func(**kwargs) return wrapper @staticmethod - def _get_ordered_policies(selected_policies=None): - """returns the ordered list of selected policies from the runtime policies""" - 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] + def get_policy_bodies(selected_policies=None): + """returns the list of policy_body objects if policy_body exists""" + if isinstance(selected_policies, dict): + return deepcopy([policy.get(POLICY_BODY) + for policy in selected_policies.values() if policy.get(POLICY_BODY)]) - @staticmethod - def get_policy_configs(): - """returns the ordered list of policy configs from the runtime policies""" - 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 get_policy_bodies(): - """returns the ordered list of policy_body objects if policy_body exists""" - return [policy.get(POLICY_BODY) - for policy in Policies._get_ordered_policies() - if policy.get(POLICY_BODY)] - - @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) + policies = ctx.instance.runtime_properties.get(POLICIES, {}) + return deepcopy([policy.get(POLICY_BODY) + for policy in policies.values() if policy.get(POLICY_BODY)]) |