3.0.0 new dataflow on policy-update and catchup

- external version 3.0.0
- internal version 5.0.0 for API and code
- changed API and functionality - new dataflow
- new dataflow between policy-handler and deployment-handler
  on policy-update and catchup
  = GET /policy - returns the deployed policy_ids+versions
    and policy-filters
  = PUT /policy updates policies on deployed components
  = new message format for policy-update and catchup
  = matching by policy_id+version and policy_filter_id only
- removed POST to /policy - the obsolete API
  = removed the 'smart' matching of the policies by policy-filter
    -- it is done by the policy-handler now

Coverage summary
Statements   : 77.06% ( 870/1129 )
Branches     : 53.55% ( 264/493 )
Functions    : 78.53% ( 150/191 )
Lines        : 77.45% ( 862/1113 )

Change-Id: I5409f32e1acd4870f1d74b466902a796fa10f6c7
Signed-off-by: Alex Shatov <alexs@att.com>
Issue-ID: DCAEGEN2-492

2 files changed, 162 insertions, 166 deletions

diff --git a/lib/cloudify.js b/lib/cloudify.js
index 2db460a..053bdb8 100644
--- a/lib/cloudify.js
+++ b/lib/cloudify.js
@@ -67,14 +67,6 @@ ExeQueue.prototype.nextExecution = function(deployment_id) {
 const exeQueue = new ExeQueue();
 exports.exeQueue = exeQueue;
 
-// Delay function--returns a promise that's resolved after 'dtime'
-// milliseconds.`
-var delay = function(dtime) {
-    return new Promise(function(resolve, reject) {
-        setTimeout(resolve, dtime);
-    });
-};
-
 // Get current status of a workflow execution
 const getExecutionStatus = function(req, execution_id) {
     var reqOptions = {
@@ -407,8 +399,8 @@ exports.getNodeInstances = function (mainReq, on_next_node_instances, offset) {
 const runQueuedExecution = function(mainReq, deployment_id, workflow_id, parameters, waitedCount) {
     mainReq = mainReq || {};
     var execution_id;
-    var exe_deployment_str = " deployment_id " + deployment_id + " to " + workflow_id
-                           + " with params(" + JSON.stringify(parameters || {}) + ")";
+    const exe_deployment_str = " deployment_id " + deployment_id + " to " + workflow_id
+                             + " with params(" + JSON.stringify(parameters || {}) + ")";
     startWorkflowExecution(mainReq, deployment_id, workflow_id, parameters)
     .then(function(result) {
         logger.info(mainReq.dcaeReqId, "result of start the execution for" + exe_deployment_str + ": " + JSON.stringify(result));
@@ -461,7 +453,7 @@ const runQueuedExecution = function(mainReq, deployment_id, workflow_id, paramet
 
 exports.executeOperation = function (mainReq, deployment_id, operation, operation_kwargs, node_instance_ids) {
     const workflow_id = "execute_operation";
-    var parameters = {
+    const parameters = {
         'operation': operation,
         'operation_kwargs': operation_kwargs,
         'node_instance_ids': node_instance_ids,
diff --git a/lib/policy.js b/lib/policy.js
index 1aefc8a..d6a701f 100644
--- a/lib/policy.js
+++ b/lib/policy.js
@@ -34,32 +34,72 @@ cloudify.setCredentials(config.cloudify.user, config.cloudify.password);
 cloudify.setLogger(logger);
 
 /**
- * receive the policy-updated message from the policy-handler
+ * send the policy-update execute-operation to cloudify per deployment
  */
-function policyUpdate(req, res, next) {
+const update_policies_on_deployments = function(result, req, policy_update) {
+    logger.info(req.dcaeReqId, "finished loading policy_deployments" + JSON.stringify(result));
+    if (result.status !== 200) {
+        const error_msg = "failed to retrieve component policies from cloudify " + result.message;
+        logger.error(createError(error_msg, result.status, "api", 502, 'cloudify-manager'), req);
+        logger.audit(req, result.status, error_msg);
+        return;
+    }
+
+    const deployment_ids = Object.keys(policy_update.policy_deployments);
+    if (!deployment_ids.length) {
+        const audit_msg = "no updated policies to apply to deployments";
+        logger.debug(req.dcaeReqId, audit_msg);
+        logger.audit(req, result.status, audit_msg);
+        return;
+    }
+    const audit_msg = "going to apply updated policies[" + Object.keys(policy_update.updated_policy_ids).length
+                    + "] and added policies[" + Object.keys(policy_update.added_policy_ids).length
+                    + "] and removed policies[" + Object.keys(policy_update.removed_policy_ids).length
+                    + "] to deployments[" + deployment_ids.length + "]";
+    logger.info(req.dcaeReqId, audit_msg + ": " + JSON.stringify(deployment_ids));
+    logger.audit(req, result.status, audit_msg);
+    deployment_ids.forEach(deployment_id => {
+        const deployment = policy_update.policy_deployments[deployment_id];
+        deployment.updated_policies = Object.keys(deployment.updated_policies).map(policy_id => {
+            return deployment.updated_policies[policy_id];
+        });
+        deployment.removed_policy_ids = Object.keys(deployment.removed_policy_ids);
+
+        logger.info(req.dcaeReqId, "ready to execute-operation policy-update on deployment " + JSON.stringify(deployment));
+        cloudify.executeOperation(req, deployment.deployment_id, POLICY_UPDATE_OPERATION,
+            {
+                'updated_policies': deployment.updated_policies,
+                'added_policies': deployment.added_policies,
+                'removed_policies': deployment.removed_policy_ids
+            },
+            deployment.node_instance_ids
+        );
+    });
+};
 
+
+/**
+ * receive the policy-updated message from the policy-handler and send to cloudify
+ *   - redesigned data-flow 2018 - R3 Casablanca
+ */
+function update_policies(req, res) {
     const policy_update = {
-        catch_up : req.body && req.body.catch_up,
         latest_policies : JSON.stringify((req.body && req.body.latest_policies) || {}),
         removed_policies : JSON.stringify((req.body && req.body.removed_policies) || {}),
-        errored_policies : JSON.stringify((req.body && req.body.errored_policies) || {}),
-        errored_scopes : JSON.stringify((req.body && req.body.errored_scopes) || []),
-        scope_prefixes : JSON.stringify((req.body && req.body.scope_prefixes) || []),
+        policy_filter_matches : JSON.stringify((req.body && req.body.policy_filter_matches) || {}),
+        policy_matches_by_filter : {},
         policy_deployments : {},
         updated_policy_ids : {},
         added_policy_ids : {},
         removed_policy_ids : {}
     };
 
-    logger.info(req.dcaeReqId, "policyUpdate "
-                + req.method + ' ' + req.protocol + '://' + req.get('host') + req.originalUrl
-                + " catch_up: " + policy_update.catch_up
-                + " latest_policies: " + policy_update.latest_policies
-                + " removed_policies: " + policy_update.removed_policies
-                + " errored_policies: " + policy_update.errored_policies
-                + " errored_scopes: " + policy_update.errored_scopes
-                + " scope_prefixes: " + policy_update.scope_prefixes
-                );
+    logger.info(req.dcaeReqId, "update_policies "
+               + req.method + ' ' + req.protocol + '://' + req.get('host') + req.originalUrl
+               + " latest_policies: " + policy_update.latest_policies
+               + " removed_policies: " + policy_update.removed_policies
+               + " policy_filter_matches: " + policy_update.policy_filter_matches
+               );
     /**
      * reply to and free up the policy_handler
      */
@@ -70,25 +110,32 @@ function policyUpdate(req, res, next) {
 
     policy_update.latest_policies = JSON.parse(policy_update.latest_policies);
     policy_update.removed_policies = JSON.parse(policy_update.removed_policies);
-    policy_update.errored_policies = JSON.parse(policy_update.errored_policies);
-    policy_update.errored_scopes = JSON.parse(policy_update.errored_scopes);
-    policy_update.scope_prefixes = JSON.parse(policy_update.scope_prefixes);
+    policy_update.policy_filter_matches = JSON.parse(policy_update.policy_filter_matches);
 
-    const is_policy_in_scopes = function(policy_id) {
-        return policy_update.scope_prefixes.some(scope_prefix => {
-            return policy_id.startsWith(scope_prefix);
+    Object.keys(policy_update.policy_filter_matches).forEach(policy_id => {
+        Object.keys(policy_update.policy_filter_matches[policy_id]).forEach(policy_filter_id => {
+            var policy_ids_by_filter = policy_update.policy_matches_by_filter[policy_filter_id];
+            if (!policy_ids_by_filter) {
+                policy_ids_by_filter = policy_update.policy_matches_by_filter[policy_filter_id] = {};
+            }
+            policy_ids_by_filter[policy_id] = true;
         });
-    };
+    });
 
-    const is_policy_in_errored_scopes = function(policy_id) {
-        return policy_update.errored_scopes.some(errored_scope => {
-            return policy_id.startsWith(errored_scope);
-        });
+    const is_policy_update_in_filters = function(policy_id, policy_filters) {
+        if (!policy_id || !policy_filters) {return null;}
+
+        const policy_update_filters = policy_update.policy_filter_matches[policy_id];
+        if (!policy_update_filters) {return null;}
+
+        return Object.keys(policy_update_filters).some(policy_filter_id =>
+            !!policy_filters[policy_filter_id]);
     };
+
     /**
      * filter out the policies to what is deployed in components and needs updating (new policyVersion)
      */
-    const collect_policy_deployments = function(node_instances) {
+    const collect_policy_deployments = function collect_policy_deployments(node_instances) {
         node_instances.forEach(node_instance => {
             if (!node_instance.runtime_properties
             || (!node_instance.runtime_properties.policies
@@ -108,17 +155,14 @@ function policyUpdate(req, res, next) {
 
             var have_policies = false;
             const deployed_policies = node_instance.runtime_properties.policies || {};
+            const deployed_policy_filters = node_instance.runtime_properties.policy_filters;
 
             Object.keys(deployed_policies).forEach(policy_id => {
                 const deployed_policy = deployed_policies[policy_id];
-                const latest_policy = policy_update.latest_policies[policy_id];
-                if (policy_update.removed_policies[policy_id]
-                  || (policy_update.catch_up
-                    && (deployed_policy.policy_body || deployment.is_deployment_busy)
-                    && !latest_policy
-                    && !policy_update.errored_policies[policy_id]
-                    && !is_policy_in_errored_scopes(policy_id)
-                    && is_policy_in_scopes(policy_id))) {
+                if ((deployed_policy.policy_body || deployment.is_deployment_busy)
+                && (policy_update.removed_policies[policy_id]
+                    || (!deployed_policy.policy_persistent
+                      && false === is_policy_update_in_filters(policy_id, deployed_policy_filters)))) {
                     have_policies = true;
                     deployment.removed_policy_ids[policy_id] = true;
                     policy_update.removed_policy_ids[policy_id] = true;
@@ -126,6 +170,7 @@ function policyUpdate(req, res, next) {
                     return;
                 }
 
+                const latest_policy = policy_update.latest_policies[policy_id];
                 if (!latest_policy || !latest_policy.policy_body
                 || isNaN(latest_policy.policy_body.policyVersion)) {return;}
 
@@ -138,86 +183,27 @@ function policyUpdate(req, res, next) {
                 logger.info(req.dcaeReqId, "going to update policy " + policy_id + " on node_instance: " + JSON.stringify(node_instance));
             });
 
-            const policy_filters = node_instance.runtime_properties.policy_filters || {};
-            const policy_filter_ids = Object.keys(policy_filters);
-            if (policy_filter_ids.length) {
-                logger.info(req.dcaeReqId, "matching latest policies to policy_filters[" + policy_filter_ids.length + "] on node_instance: " + JSON.stringify(node_instance));
-                try {
-                    Object.keys(policy_update.latest_policies).forEach(policy_id => {
-                        if (!deployment.is_deployment_busy && deployed_policies[policy_id]) {return;}
-
-                        const latest_policy = policy_update.latest_policies[policy_id];
-                        const policy_body = latest_policy && latest_policy.policy_body;
-                        if (!policy_body || isNaN(policy_body.policyVersion)) {return;}
-                        const policy_name = policy_body.policyName;
-                        if (!policy_name) {return;}
-                        const matching_conditions = policy_body.matchingConditions || {};
-
-                        logger.debug(req.dcaeReqId, "matching policy " + JSON.stringify(latest_policy));
-                        policy_filter_ids.some(policy_filter_id => {
-                            const policy_filter = policy_filters[policy_filter_id].policy_filter;
-                            if (!policy_filter || !policy_filter.policyName) {return false;}
-
-                            logger.debug(req.dcaeReqId, "matching to policy_filter " + JSON.stringify(policy_filter));
-
-                            if (!!policy_filter.onapName
-                            && policy_filter.onapName !== matching_conditions.ONAPName) {
-                                logger.debug(req.dcaeReqId, "not match policy_filter_id " + policy_filter_id
-                                    + " by ONAPName: "
-                                    + policy_filter.onapName + " !== " + matching_conditions.ONAPName);
-                                return false;
-                            }
-                            if (!!policy_filter.configName
-                            && policy_filter.configName !== matching_conditions.ConfigName) {
-                                logger.debug(req.dcaeReqId, "not match policy_filter_id " + policy_filter_id
-                                    + " by configName: "
-                                    + policy_filter.configName + " !== " + matching_conditions.ConfigName);
-                                return false;
-                            }
-
-                            if (policy_filter.configAttributes
-                            && !Object.keys(policy_filter.configAttributes).every(filter_key => {
-                                    return (matching_conditions.hasOwnProperty(filter_key)
-                                        && policy_filter.configAttributes[filter_key]
-                                       === matching_conditions[filter_key]);
-                                })) {
-                                logger.debug(req.dcaeReqId, "not match policy_filter_id " + policy_filter_id
-                                    + " by configAttributes: "
-                                    + JSON.stringify(policy_filter.configAttributes) + " !== " + JSON.stringify(matching_conditions));
-                                return false;
-                            }
-
-                            if (policy_filter.policyName !== policy_id && policy_filter.policyName !== policy_name) {
-                                const match_policy_name = new RegExp(policy_filter.policyName);
-                                if (!match_policy_name.test(policy_name)) {
-                                    logger.debug(req.dcaeReqId, "not match policy_filter_id " + policy_filter_id
-                                        + " by policyName: "
-                                        + policy_filter.policyName + " versus " + policy_name);
-                                    return false;
-                                }
-                            }
-
-                            have_policies = true;
-                            if (!deployment.added_policies[policy_filter_id]) {
-                                deployment.added_policies[policy_filter_id] = {
-                                    "policy_filter_id" : policy_filter_id,
-                                    "policies" : {}
-                                };
-                            }
-                            deployment.added_policies[policy_filter_id].policies[policy_id] = latest_policy;
-                            policy_update.added_policy_ids[policy_id] = true;
-                            logger.info(req.dcaeReqId, "going to add policy " + JSON.stringify(latest_policy)
-                                + " per policy_filter_id " + policy_filter_id
-                                + " on node_instance: " + JSON.stringify(node_instance));
-                            return true;
-                        });
-                    });
-                } catch (e) {
-                    const error_msg = "error on matching policy to filter " + (e.message || "")
-                        + " " + (e.stack || "").replace(/\n/g, " ")
-                    logger.error(createError(error_msg, 500, "api", 553, 'deployment-handler'), req);
-                }
-            }
+            Object.keys(deployed_policy_filters || {}).forEach(policy_filter_id => {
+                Object.keys(policy_update.policy_matches_by_filter[policy_filter_id] || {}).forEach(policy_id => {
+                    if (!deployment.is_deployment_busy && deployed_policies[policy_id]) {return;}
+
+                    const latest_policy = policy_update.latest_policies[policy_id];
+                    const policy_body = latest_policy && latest_policy.policy_body;
+                    if (!policy_body || isNaN(policy_body.policyVersion)) {return;}
+
+                    have_policies = true;
+                    deployment.added_policies[policy_filter_id] = deployment.added_policies[policy_filter_id] || {
+                        "policy_filter_id" : policy_filter_id,
+                        "policies" : {}
+                    };
+
+                    deployment.added_policies[policy_filter_id].policies[policy_id] = latest_policy;
+                    policy_update.added_policy_ids[policy_id] = true;
+                    logger.info(req.dcaeReqId, "going to add policy " + JSON.stringify(latest_policy)
+                        + " per policy_filter_id " + policy_filter_id
+                        + " on node_instance: " + JSON.stringify(node_instance));
+                });
+            });
 
             if (have_policies) {
                 deployment.node_instance_ids.push(node_instance.id);
@@ -228,50 +214,66 @@ function policyUpdate(req, res, next) {
         logger.info(req.dcaeReqId, "collected policy_deployments to update " + JSON.stringify(policy_update.policy_deployments));
     };
 
-    const update_policies_on_deployments = function(result) {
-        logger.info(req.dcaeReqId, "finished loading policy_deployments" + JSON.stringify(result));
-        if (result.status !== 200) {
-            const error_msg = "failed to retrieve component policies from cloudify " + result.message;
-            logger.error(createError(error_msg, result.status, "api", 502, 'cloudify-manager'), req);
-            logger.audit(req, result.status, error_msg);
-            return;
-        }
+    cloudify.getNodeInstances(req, collect_policy_deployments)
+            .then(result => {update_policies_on_deployments(result, req, policy_update);});
+}
 
-        const deployment_ids = Object.keys(policy_update.policy_deployments);
-        if (!deployment_ids.length) {
-            const audit_msg = "no updated policies to apply to deployments";
-            logger.debug(req.dcaeReqId, audit_msg);
-            logger.audit(req, result.status, audit_msg);
-            return;
-        }
-        const audit_msg = "going to apply updated policies[" + Object.keys(policy_update.updated_policy_ids).length
-                        + "] and added policies[" + Object.keys(policy_update.added_policy_ids).length
-                        + "] and removed policies[" + Object.keys(policy_update.removed_policy_ids).length
-                        + "] to deployments[" + deployment_ids.length + "]";
-        logger.info(req.dcaeReqId, audit_msg + ": " + JSON.stringify(deployment_ids));
-        logger.audit(req, result.status, audit_msg);
-        deployment_ids.forEach(deployment_id => {
-            const deployment = policy_update.policy_deployments[deployment_id];
-            deployment.updated_policies = Object.keys(deployment.updated_policies).map(policy_id => {
-                return deployment.updated_policies[policy_id];
-            });
-            deployment.removed_policy_ids = Object.keys(deployment.removed_policy_ids);
-
-            logger.info(req.dcaeReqId, "ready to execute-operation policy-update on deployment " + JSON.stringify(deployment));
-            cloudify.executeOperation(req, deployment.deployment_id, POLICY_UPDATE_OPERATION,
-                {
-                    'updated_policies': deployment.updated_policies,
-                    'added_policies': deployment.added_policies,
-                    'removed_policies': deployment.removed_policy_ids
-                },
-                deployment.node_instance_ids
-            );
+/**
+ * retrieve the unique set of policies and policy-filters from cloudify
+ */
+function get_policies_from_cloudify(req, res, next) {
+    logger.info(req.dcaeReqId, "getPoliciesFromCloudify " + req.originalUrl);
+    const response = {"requestID": req.dcaeReqId};
+    response.started = new Date();
+    response.server_instance_uuid = process.mainModule.exports.config.server_instance_uuid;
+    response.policies = {};
+    response.policy_filters = {};
+
+    cloudify.getNodeInstances(req, function(node_instances) {
+        node_instances.forEach(node_instance => {
+            if (!node_instance.runtime_properties) {return;}
+            const pending_update = cloudify.exeQueue.isDeploymentBusy(node_instance.deployment_id);
+
+            if (node_instance.runtime_properties.policies) {
+                Object.keys(node_instance.runtime_properties.policies).forEach(policy_id => {
+                    const deployed_policy = response.policies[policy_id] || {
+                        "policy_id": policy_id,
+                        "policy_versions": {}
+                    };
+                    const policy = node_instance.runtime_properties.policies[policy_id];
+                    if (policy.policy_body && policy.policy_body.policyVersion) {
+                        deployed_policy.policy_versions[policy.policy_body.policyVersion] = true;
+                    }
+                    deployed_policy.pending_update = deployed_policy.pending_update || pending_update;
+                    response.policies[policy_id] = deployed_policy;
+                });
+            }
+            if (node_instance.runtime_properties.policy_filters) {
+                Object.keys(node_instance.runtime_properties.policy_filters).forEach(policy_filter_id => {
+                    node_instance.runtime_properties.policy_filters[policy_filter_id].pending_update = pending_update;
+                });
+                Object.assign(response.policy_filters, node_instance.runtime_properties.policy_filters);
+            }
         });
-    };
 
-    cloudify.getNodeInstances(req, collect_policy_deployments).then(update_policies_on_deployments);
+        logger.info(req.dcaeReqId, "deployed policies: " + JSON.stringify(response.policies)
+                                 + " policy_filters: " + JSON.stringify(response.policy_filters)
+        );
+    })
+    .then(function(result) {
+        response.ended = new Date();
+        response.status = result.status;
+        response.message = result.message;
+        logger.info(req.dcaeReqId, result.message);
+        if (result.status !== 200) {
+            logger.error(createError(result.message, result.status, "api", 502, 'cloudify-manager'), req);
+        }
+        res.status(result.status).json(response);
+        logger.audit(req, result.status, result.message);
+    });
 }
 
+
 /**
  * retrieve all component-policies from cloudify
  */
@@ -358,7 +360,9 @@ app.use(function(req, res, next) {
 	next();
 });
 
-app.post('/', policyUpdate);
+app.get('/', get_policies_from_cloudify);
+app.put('/', update_policies);
+
 app.get('/components', getComponentPoliciesFromCloudify);
 
 module.exports = app;