diff options
Diffstat (limited to 'onap-dcae-dcaepolicy-lib/onap_dcae_dcaepolicy_lib')
3 files changed, 235 insertions, 36 deletions
diff --git a/onap-dcae-dcaepolicy-lib/onap_dcae_dcaepolicy_lib/__init__.py b/onap-dcae-dcaepolicy-lib/onap_dcae_dcaepolicy_lib/__init__.py index 18b1b0c..547b16f 100644 --- a/onap-dcae-dcaepolicy-lib/onap_dcae_dcaepolicy_lib/__init__.py +++ b/onap-dcae-dcaepolicy-lib/onap_dcae_dcaepolicy_lib/__init__.py @@ -19,4 +19,4 @@ """expose the Policies class on the package level""" -from .dcae_policy import Policies, POLICIES, POLICY_MESSAGE_TYPE +from .dcae_policy import Policies 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 e4c5cea..b064157 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 @@ -20,14 +20,15 @@ """dcae_policy contains decorators for the policy lifecycle in cloudify""" import json +import traceback 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 .policies_output import PoliciesOutput from .utils import Utils POLICIES = 'policies' @@ -47,12 +48,14 @@ POLICY_FILTER = 'policy_filter' POLICY_FILTER_ID = 'policy_filter_id' POLICY_TYPE = 'policyType' POLICY_TYPE_MICROSERVICE = 'MicroService' -POLICY_CONFIG_MS = "Config_MS" +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' + +ACTION_GATHERED = "gathered" +ACTION_UPDATED = "updated" class Policies(object): """static class for policy operations""" @@ -195,15 +198,15 @@ class Policies(object): """ policy_defaulted_fields = ctx.instance.runtime_properties.get(POLICY_DEFAULTED_FIELDS, {}) policy_defaulted_fields.update( - (k, True) + (field_name, True) for policy in Policies._removed_policies.itervalues() - for k in Policies._get_config_from_policy(policy) or {} + for field_name 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] + 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 @@ -230,13 +233,25 @@ class Policies(object): Policies._set_policy_apply_order(policies) @staticmethod + def _get_policy_bodies_dict(policies): + """returns a dict of policy_id -> policy_body if policy_body exists""" + if not policies: + return {} + + return dict((policy_id, policy.get(POLICY_BODY)) + for policy_id, policy in policies.iteritems() if policy.get(POLICY_BODY) + ) + + @staticmethod def gather_policies_to_node(policy_apply_order_path=None): """ - decorate with @Policies.gather_policies_to_node to + 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"]. + Stores <scn>:policies data in consul-kv + Call Policies.calc_latest_application_config() to apply policies onto application_config. """ def gather_policies_decorator(func): @@ -301,6 +316,8 @@ class Policies(object): if ctx.type != NODE_INSTANCE: raise NonRecoverableError("can only invoke gather_policies_to_node on node") + policies_outputted = False + policy_bodies = [] try: Policies._store_policy_apply_order_clause(policy_apply_order_path) @@ -313,12 +330,21 @@ class Policies(object): Policies._set_policies(policies) if policy_filters: ctx.instance.runtime_properties[POLICY_FILTERS] = policy_filters + + policy_bodies = Policies._get_policy_bodies_dict(policies) + if policy_bodies: + policies_outputted = PoliciesOutput.store_policies(ACTION_GATHERED, policy_bodies) 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) + func_result = func(*args, **kwargs) + + if not policies_outputted and policy_bodies: + PoliciesOutput.store_policies(ACTION_GATHERED, policy_bodies) + + return func_result return wrapper return gather_policies_decorator @@ -349,19 +375,17 @@ class Policies(object): for policy_id in removed_policies: removed_policy = policies.get(policy_id) - if removed_policy: + if removed_policy and POLICY_BODY in 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: + if removed_policy.get(POLICY_PERSISTENT): del policies[policy_id][POLICY_BODY] + else: + del policies[policy_id] - new_policies = dict((p_id, policy) + new_policies = dict((policy_id, policy) for policy_filter_id in policy_filters - for (p_id, policy) in added_policies - .get(policy_filter_id, {}) - .get(POLICIES, {}) - .iteritems()) + for (policy_id, policy) in added_policies.get(policy_filter_id, {}) + .get(POLICIES, {}).iteritems()) ctx.logger.info("new_policies: {0}".format(json.dumps(new_policies))) @@ -373,13 +397,17 @@ class Policies(object): continue updated_policies.append(policy) - skipped = {"ignored" : [], "unexpected" : [], "same" : []} + skipped = {"ignored": [], "unexpected": [], "same": [], "duplicate": []} 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 + if policy_id in Policies._updated_policies: + skipped["duplicate"].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: @@ -395,20 +423,25 @@ class Policies(object): 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 skipped["same"] or skipped["ignored"] or skipped["unexpected"] or skipped["duplicate"]: + ctx.logger.info("skipped updates on policies: {0}".format(json.dumps(skipped))) if Policies._updated_policies or Policies._removed_policies: Policies._set_policies(policies) + policy_bodies = Policies._get_policy_bodies_dict(policies) + PoliciesOutput.store_policies(ACTION_UPDATED, policy_bodies) @staticmethod - def update_policies_on_node(configs_only=True): - """decorate each policy_update operation with @Policies.update_policies_on_node to + def update_policies_on_node(configs_only=False): + """ + 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 object (False) + instead of the whole policy_body object (False) Passes through the filtered list of updated_policies that apply to the current node instance @@ -439,10 +472,16 @@ class Policies(object): if configs_only: updated_policies = Utils.remove_empties( [Policies._get_config_from_policy(policy) - for policy in updated_policies or []] + for policy in updated_policies] ) removed_policies = [Policies._get_config_from_policy(policy) - for policy in removed_policies or []] + 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] except Exception as ex: error = "Failed to update the policies {0}".format(str(ex)) @@ -450,16 +489,41 @@ class Policies(object): raise NonRecoverableError(error) if updated_policies or removed_policies: - return func(updated_policies, removed_policies=removed_policies, **kwargs) + return func( + updated_policies, + removed_policies=removed_policies, + policies=Policies.get_policy_bodies(), + **kwargs) return wrapper return update_policies_decorator @staticmethod - def _get_ordered_policies(selected_policies=None): + def cleanup_policies_on_node(func): """ - returns the ordered list of selected policies from the runtime policies - ordered by policy_id values + decorate each policy_update operation with @Policies.cleanup_policies_on_node to + remove <scn>:policies data in consul-kv """ + if not func: + return + + @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())) + + 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: @@ -475,10 +539,7 @@ class Policies(object): @staticmethod def get_policy_configs(): - """ - returns the list of policy configs from the runtime policies - ordered by policy_id values - """ + """returns the ordered list of policy configs from the runtime policies""" if ctx.type != NODE_INSTANCE: return [] @@ -489,6 +550,13 @@ class Policies(object): ) @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. diff --git a/onap-dcae-dcaepolicy-lib/onap_dcae_dcaepolicy_lib/policies_output.py b/onap-dcae-dcaepolicy-lib/onap_dcae_dcaepolicy_lib/policies_output.py new file mode 100644 index 0000000..7a1a8e8 --- /dev/null +++ b/onap-dcae-dcaepolicy-lib/onap_dcae_dcaepolicy_lib/policies_output.py @@ -0,0 +1,131 @@ +# org.onap.dcae +# ================================================================================ +# Copyright (c) 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. +"""client to talk to consul at the standard port 8500 on localhost""" + +import base64 +import json +import urllib +import uuid +from datetime import datetime + +import requests +from cloudify import ctx + + +class PoliciesOutput(object): + """static class for store-delete policies in consul kv""" + # it is safe to assume that consul agent is at localhost:8500 along with cloudify manager + CONSUL_TRANSACTION_URL = "http://localhost:8500/v1/txn" + + POLICIES_EVENT = 'policies_event' + + POLICIES_FOLDER_MASK = "{0}:policies/{1}" + MAX_OPS_PER_TXN = 64 + # MAX_VALUE_LEN = 512 * 1000 + + OPERATION_SET = "set" + OPERATION_DELETE = "delete" + OPERATION_DELETE_FOLDER = "delete-tree" + SERVICE_COMPONENT_NAME = "service_component_name" + + @staticmethod + def _gen_txn_operation(verb, service_component_name, key=None, value=None): + """returns the properly formatted operation to be used inside transaction""" + key = PoliciesOutput.POLICIES_FOLDER_MASK.format(service_component_name, urllib.quote(key or "")) + if value: + return {"KV": {"Verb": verb, "Key": key, "Value": base64.b64encode(value)}} + return {"KV": {"Verb": verb, "Key": key}} + + @staticmethod + def _run_transaction(operation_name, txn): + """run a single transaction of several operations at consul /txn""" + if not txn: + return + + response = None + try: + response = requests.put(PoliciesOutput.CONSUL_TRANSACTION_URL, json=txn) + except requests.exceptions.RequestException as ex: + ctx.logger.error("failed to {0} at {1}: {2} on txn={3}" + .format(operation_name, PoliciesOutput.CONSUL_TRANSACTION_URL, + str(ex), json.dumps(txn))) + return + + if response.status_code != requests.codes.ok: + ctx.logger.error("failed {0} {1}: {2} text={3} txn={4} headers={5}" + .format(operation_name, PoliciesOutput.CONSUL_TRANSACTION_URL, response.status_code, + response.text, json.dumps(txn), + json.dumps(dict(response.request.headers.items())))) + return + ctx.logger.info("response for {0} {1}: {2} text={3} txn={4} headers={5}" + .format(operation_name, PoliciesOutput.CONSUL_TRANSACTION_URL, response.status_code, + response.text, json.dumps(txn), + json.dumps(dict(response.request.headers.items())))) + return True + + @staticmethod + def store_policies(action, policy_bodies): + """put the policy_bodies for service_component_name into consul-kv""" + service_component_name = ctx.instance.runtime_properties.get(PoliciesOutput.SERVICE_COMPONENT_NAME) + if not service_component_name: + ctx.logger.warn("failed to find service_component_name to store_policies in consul-kv") + return False + + event = { + "action": action, + "timestamp": (datetime.utcnow().isoformat()[:-3] + 'Z'), + "update_id": str(uuid.uuid4()), + "policies_count": len(policy_bodies) + } + ctx.instance.runtime_properties[PoliciesOutput.POLICIES_EVENT] = event + + store_policies = [ + PoliciesOutput._gen_txn_operation(PoliciesOutput.OPERATION_SET, service_component_name, + "items/" + policy_id, json.dumps(policy_body)) + for policy_id, policy_body in policy_bodies.iteritems() + ] + txn = [ + PoliciesOutput._gen_txn_operation(PoliciesOutput.OPERATION_DELETE_FOLDER, service_component_name), + PoliciesOutput._gen_txn_operation(PoliciesOutput.OPERATION_SET, service_component_name, "event", json.dumps(event)) + ] + idx_step = PoliciesOutput.MAX_OPS_PER_TXN - len(txn) + for idx in xrange(0, len(store_policies), idx_step): + txn += store_policies[idx : idx + idx_step] + if not PoliciesOutput._run_transaction("store_policies", txn): + return False + txn = [] + + PoliciesOutput._run_transaction("store_policies", txn) + return True + + @staticmethod + def delete_policies(): + """delete policies for service_component_name in consul-kv""" + if PoliciesOutput.POLICIES_EVENT not in ctx.instance.runtime_properties: + return + + service_component_name = ctx.instance.runtime_properties.get(PoliciesOutput.SERVICE_COMPONENT_NAME) + if not service_component_name: + ctx.logger.warn("failed to find service_component_name to delete_policies in consul-kv") + return + + delete_policies = [ + PoliciesOutput._gen_txn_operation(PoliciesOutput.OPERATION_DELETE_FOLDER, service_component_name) + ] + PoliciesOutput._run_transaction("delete_policies", delete_policies) |