policy-handler seed code

Change-Id: I35cd80b6e082f4b84740bab752774e8abc40ca35
Issue-Id: DCAEGEN2-46
Signed-off-by: Alex Shatov <alexs@att.com>

1 files changed, 342 insertions, 0 deletions

diff --git a/policyhandler/policy_rest.py b/policyhandler/policy_rest.py
new file mode 100644
index 0000000..d49164c
--- /dev/null
+++ b/policyhandler/policy_rest.py
@@ -0,0 +1,342 @@
+"""policy-client communicates with policy-engine thru REST API"""
+
+# org.onap.dcae
+# ================================================================================
+# Copyright (c) 2017 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.
+
+import logging
+import json
+import copy
+import re
+import time
+from multiprocessing.dummy import Pool as ThreadPool
+import requests
+
+from .config import Config
+from .policy_consts import POLICY_ID, POLICY_VERSION, POLICY_NAME, POLICY_GET_CONFIG, \
+    POLICY_BODY, POLICY_CONFIG
+from .onap.audit import REQUEST_X_ECOMP_REQUESTID, Audit, AuditHttpCode, AuditResponseCode
+
+class PolicyUtils(object):
+    """policy-client utils"""
+    _logger = logging.getLogger("policy_handler.policy_utils")
+    _policy_name_ext = re.compile('[.][0-9]+[.][a-zA-Z]+$')
+
+    @staticmethod
+    def safe_json_parse(json_str):
+        """try parsing json without exception - returns the json_str back if fails"""
+        if not json_str:
+            return json_str
+        try:
+            return json.loads(json_str)
+        except ValueError as err:
+            PolicyUtils._logger.warn("unexpected json %s: %s", str(json_str), str(err))
+        return json_str
+
+    @staticmethod
+    def extract_policy_id(policy_name):
+        """ policy_name  = policy_id + "." + <version> + "." + <extension>
+        For instance,
+        policy_name      = DCAE_alex.Config_alex_policy_number_1.3.xml
+               policy_id = DCAE_alex.Config_alex_policy_number_1
+            policy_scope = DCAE_alex
+            policy_class = Config
+          policy_version = 3
+        type = extension = xml
+               delimiter = "."
+        policy_class_delimiter = "_"
+        policy_name in PAP = DCAE_alex.alex_policy_number_1
+        """
+        if not policy_name:
+            return
+        return PolicyUtils._policy_name_ext.sub('', policy_name)
+
+    @staticmethod
+    def parse_policy_config(policy):
+        """try parsing the config in policy."""
+        if policy and POLICY_BODY in policy and POLICY_CONFIG in policy[POLICY_BODY]:
+            policy[POLICY_BODY][POLICY_CONFIG] = PolicyUtils.safe_json_parse(
+                policy[POLICY_BODY][POLICY_CONFIG])
+        return policy
+
+    @staticmethod
+    def convert_to_policy(policy_config):
+        """wrap policy_config received from policy-engine with policy_id."""
+        if not policy_config or POLICY_NAME not in policy_config \
+        or POLICY_VERSION not in policy_config or not policy_config[POLICY_VERSION]:
+            return
+        policy_id = PolicyUtils.extract_policy_id(policy_config[POLICY_NAME])
+        if not policy_id:
+            return
+        return {POLICY_ID:policy_id, POLICY_BODY:policy_config}
+
+    @staticmethod
+    def select_latest_policy(policy_configs):
+        """For some reason, the policy-engine returns all version of the policy_configs.
+        DCAE-Controller is only interested in the latest version
+        """
+        if not policy_configs:
+            return
+        latest_policy_config = {}
+        for policy_config in policy_configs:
+            if POLICY_VERSION not in policy_config or not policy_config[POLICY_VERSION] \
+            or not policy_config[POLICY_VERSION].isdigit():
+                continue
+            if not latest_policy_config \
+                or int(policy_config[POLICY_VERSION]) \
+                 > int(latest_policy_config[POLICY_VERSION]):
+                latest_policy_config = policy_config
+
+        return PolicyUtils.parse_policy_config(PolicyUtils.convert_to_policy(latest_policy_config))
+
+    @staticmethod
+    def select_latest_policies(policy_configs):
+        """For some reason, the policy-engine returns all version of the policy_configs.
+        DCAE-Controller is only interested in the latest versions
+        """
+        if not policy_configs:
+            return {}
+        policies = {}
+        for policy_config in policy_configs:
+            policy = PolicyUtils.convert_to_policy(policy_config)
+            if not policy or POLICY_ID not in policy or POLICY_BODY not in policy:
+                continue
+            if POLICY_VERSION not in policy[POLICY_BODY] \
+            or not policy[POLICY_BODY][POLICY_VERSION] \
+            or not policy[POLICY_BODY][POLICY_VERSION].isdigit():
+                continue
+            if policy[POLICY_ID] not in policies:
+                policies[policy[POLICY_ID]] = policy
+                continue
+            if int(policy[POLICY_BODY][POLICY_VERSION]) \
+             > int(policies[policy[POLICY_ID]][POLICY_BODY][POLICY_VERSION]):
+                policies[policy[POLICY_ID]] = policy
+
+        for policy_id in policies:
+            policies[policy_id] = PolicyUtils.parse_policy_config(policies[policy_id])
+
+        return policies
+
+class PolicyRest(object):
+    """ policy-engine """
+    _logger = logging.getLogger("policy_handler.policy_rest")
+    _lazy_inited = False
+    _url = None
+    _headers = None
+    _target_entity = None
+    _thread_pool_size = 4
+    _scope_prefixes = None
+    _scope_thread_pool_size = 4
+    _policy_retry_count = 1
+    _policy_retry_sleep = 0
+
+    @staticmethod
+    def _lazy_init():
+        """init static config"""
+        if PolicyRest._lazy_inited:
+            return
+        PolicyRest._lazy_inited = True
+        config = Config.config[Config.FIELD_POLICY_ENGINE]
+        PolicyRest._url = config["url"] + config["path_api"]
+        PolicyRest._headers = config["headers"]
+        PolicyRest._target_entity = config.get("target_entity", Config.FIELD_POLICY_ENGINE)
+        PolicyRest._thread_pool_size = Config.config.get("thread_pool_size", 4)
+        if PolicyRest._thread_pool_size < 2:
+            PolicyRest._thread_pool_size = 2
+        PolicyRest._scope_prefixes = Config.config["scope_prefixes"]
+        PolicyRest._scope_thread_pool_size = min(PolicyRest._thread_pool_size, \
+                                             len(PolicyRest._scope_prefixes))
+
+        PolicyRest._policy_retry_count = Config.config.get("policy_retry_count", 1) or 1
+        PolicyRest._policy_retry_sleep = Config.config.get("policy_retry_sleep", 0)
+
+        PolicyRest._logger.info("PolicyClient url(%s) headers(%s) scope-prefixes(%s)", \
+            PolicyRest._url, Audit.log_json_dumps(PolicyRest._headers), \
+            json.dumps(PolicyRest._scope_prefixes))
+
+    @staticmethod
+    def _post(audit, path, json_body):
+        """Communication with the policy-engine"""
+        full_path = PolicyRest._url + path
+        sub_aud = Audit(aud_parent=audit, targetEntity=PolicyRest._target_entity, \
+            targetServiceName=full_path)
+
+        msg = json.dumps(json_body)
+        headers = copy.copy(PolicyRest._headers)
+        headers[REQUEST_X_ECOMP_REQUESTID] = sub_aud.request_id
+        headers_str = Audit.log_json_dumps(headers)
+
+        log_line = "post to PDP {0} msg={1} headers={2}".format(full_path, msg, headers_str)
+        sub_aud.metrics_start(log_line)
+        PolicyRest._logger.info(log_line)
+        res = None
+        try:
+            res = requests.post(full_path, json=json_body, headers=headers)
+        except requests.exceptions.RequestException as ex:
+            error_code = AuditHttpCode.SERVICE_UNAVAILABLE_ERROR.value
+            error_msg = "failed to post to PDP {0} {1} msg={2} headers={3}" \
+                .format(full_path, str(ex), msg, headers_str)
+
+            PolicyRest._logger.exception(error_msg)
+            sub_aud.set_http_status_code(error_code)
+            audit.set_http_status_code(error_code)
+            sub_aud.metrics(error_msg)
+            return (error_code, None)
+
+        log_line = "response from PDP to post {0}: {1} msg={2} text={3} headers={4}".format( \
+            full_path, res.status_code, msg, res.text, \
+            Audit.log_json_dumps(dict(res.request.headers.items())))
+        sub_aud.set_http_status_code(res.status_code)
+        sub_aud.metrics(log_line)
+        PolicyRest._logger.info(log_line)
+
+        if res.status_code == requests.codes.ok:
+            return res.status_code, res.json()
+
+        return res.status_code, None
+
+    @staticmethod
+    def get_latest_policy(aud_policy_name):
+        """Get the latest policy for the policy_name from the policy-engine"""
+        PolicyRest._lazy_init()
+        audit, policy_name = aud_policy_name
+
+        status_code = 0
+        latest_policy = None
+        for retry in xrange(1, PolicyRest._policy_retry_count + 1):
+            PolicyRest._logger.debug("%s", policy_name)
+            status_code, policy_configs = PolicyRest._post(audit, POLICY_GET_CONFIG, \
+                                            {POLICY_NAME:policy_name})
+            PolicyRest._logger.debug("%s %s policy_configs: %s", status_code, policy_name, \
+                        json.dumps(policy_configs or []))
+            latest_policy = PolicyUtils.select_latest_policy(policy_configs)
+            if not latest_policy:
+                audit.error("received unexpected policy data from PDP for policy_name={0}: {1}" \
+                    .format(policy_name, json.dumps(policy_configs or [])), \
+                    errorCode=AuditResponseCode.DATA_ERROR.value, \
+                    errorDescription=AuditResponseCode.get_human_text( \
+                        AuditResponseCode.DATA_ERROR))
+
+            if latest_policy or not audit.retry_get_config \
+            or not PolicyRest._policy_retry_sleep \
+            or AuditResponseCode.PERMISSION_ERROR.value \
+            == AuditResponseCode.get_response_code(status_code).value:
+                break
+
+            if retry == PolicyRest._policy_retry_count:
+                audit.warn("gave up retrying {0} from PDP after #{1} for policy_name={2}" \
+                    .format(POLICY_GET_CONFIG, retry, policy_name), \
+                    errorCode=AuditResponseCode.DATA_ERROR.value, \
+                    errorDescription=AuditResponseCode.get_human_text( \
+                            AuditResponseCode.DATA_ERROR))
+                break
+
+            audit.warn("retry #{0} {1} from PDP in {2} secs for policy_name={3}" \
+                .format(retry, POLICY_GET_CONFIG, PolicyRest._policy_retry_sleep, policy_name), \
+                errorCode=AuditResponseCode.DATA_ERROR.value, \
+                errorDescription=AuditResponseCode.get_human_text( \
+                        AuditResponseCode.DATA_ERROR))
+            time.sleep(PolicyRest._policy_retry_sleep)
+
+        audit.set_http_status_code(status_code)
+        if not latest_policy:
+            audit.set_http_status_code(AuditHttpCode.DATA_NOT_FOUND_ERROR.value)
+        return latest_policy
+
+    @staticmethod
+    def get_latest_policies_by_names(aud_policy_names):
+        """Get the latest policies of the list of policy_names from the policy-engine"""
+        PolicyRest._lazy_init()
+        audit, policy_names = aud_policy_names
+        if not policy_names:
+            return
+
+        audit.metrics_start("get_latest_policies_by_names {0} {1}".format( \
+            len(policy_names), json.dumps(policy_names)))
+        PolicyRest._logger.debug("%d %s", len(policy_names), json.dumps(policy_names))
+
+        thread_count = min(PolicyRest._thread_pool_size, len(policy_names))
+        apns = [(audit, policy_name) for policy_name in policy_names]
+        policies = None
+        if thread_count == 1:
+            policies = [PolicyRest.get_latest_policy(apns[0])]
+        else:
+            pool = ThreadPool(thread_count)
+            policies = pool.map(PolicyRest.get_latest_policy, apns)
+            pool.close()
+            pool.join()
+
+        audit.metrics("result get_latest_policies_by_names {0} {1}: {2} {3}".format( \
+            len(policy_names), json.dumps(policy_names), len(policies), json.dumps(policies)), \
+            targetEntity=PolicyRest._target_entity, targetServiceName=POLICY_GET_CONFIG)
+        policies = dict([(policy[POLICY_ID], policy) \
+            for policy in policies if policy and POLICY_ID in policy])
+        PolicyRest._logger.debug("policies %s", json.dumps(policies))
+        if not policies:
+            audit.set_http_status_code(AuditHttpCode.DATA_NOT_FOUND_ERROR.value)
+        return policies
+
+    @staticmethod
+    def _get_latest_policies(aud_scope_prefix):
+        """Get the latest policies of the same scope from the policy-engine"""
+        audit, scope_prefix = aud_scope_prefix
+        PolicyRest._logger.debug("%s", scope_prefix)
+        status_code, policy_configs = PolicyRest._post(audit, POLICY_GET_CONFIG, \
+                                          {POLICY_NAME:scope_prefix + ".*"})
+        audit.set_http_status_code(status_code)
+        PolicyRest._logger.debug("%s policy_configs: %s %s", status_code, \
+              scope_prefix, json.dumps(policy_configs or []))
+        latest_policies = PolicyUtils.select_latest_policies(policy_configs)
+
+        if not latest_policies:
+            audit.set_http_status_code(AuditHttpCode.DATA_NOT_FOUND_ERROR.value)
+            audit.error("received unexpected policies data from PDP for scope {0}: {1}".format( \
+                scope_prefix, json.dumps(policy_configs or [])), \
+                errorCode=AuditResponseCode.DATA_ERROR.value, \
+                errorDescription=AuditResponseCode.get_human_text( \
+                        AuditResponseCode.DATA_ERROR))
+        return latest_policies
+
+    @staticmethod
+    def get_latest_policies(audit):
+        """Get the latest policies of the same scope from the policy-engine"""
+        PolicyRest._lazy_init()
+        PolicyRest._logger.debug("%s", json.dumps(PolicyRest._scope_prefixes))
+
+        audit.metrics_start("get_latest_policies for scopes {0} {1}".format( \
+            len(PolicyRest._scope_prefixes), json.dumps(PolicyRest._scope_prefixes)))
+        asps = [(audit, scope_prefix) for scope_prefix in PolicyRest._scope_prefixes]
+        latest_policies = None
+        if PolicyRest._scope_thread_pool_size == 1:
+            latest_policies = [PolicyRest._get_latest_policies(asps[0])]
+        else:
+            pool = ThreadPool(PolicyRest._scope_thread_pool_size)
+            latest_policies = pool.map(PolicyRest._get_latest_policies, asps)
+            pool.close()
+            pool.join()
+
+        audit.metrics("total result get_latest_policies for scopes {0} {1}: {2} {3}".format( \
+            len(PolicyRest._scope_prefixes), json.dumps(PolicyRest._scope_prefixes), \
+            len(latest_policies), json.dumps(latest_policies)), \
+            targetEntity=PolicyRest._target_entity, targetServiceName=POLICY_GET_CONFIG)
+
+        latest_policies = dict(pair for lp in latest_policies if lp for pair in lp.items())
+        PolicyRest._logger.debug("latest_policies: %s %s", \
+              json.dumps(PolicyRest._scope_prefixes), json.dumps(latest_policies))
+
+        return latest_policies