allow access entities using NGSIv2 API for non_signal rules

bin/perseo

@@ -107,7 +107,8 @@ function loadConfiguration() {
         'PERSEO_AUTHENTICATION_USER',
         'PERSEO_AUTHENTICATION_PASSWORD',
         'PERSEO_AUTHENTICATION_SERVICE',
-        'PERSEO_MAX_RULES_BY_CORR'
+        'PERSEO_MAX_RULES_BY_CORR',
+        'PERSEO_CHECK_NON_SIGNAL_BY_API'
     ];
 
     const protectedVariables = [
@@ -234,6 +235,9 @@ function loadConfiguration() {
     if (process.env.PERSEO_MAX_RULES_BY_CORR) {
         config.maxRulesByCorr = process.env.PERSEO_MAX_RULES_BY_CORR;
     }
+    if (process.env.PERSEO_CHECK_NON_SIGNAL_BY_API) {
+        config.nonSignalByAPI = process.env.PERSEO_CHECK_NON_SIGNAL_BY_API;
+    }
 }
 
 loadConfiguration();

config.js

@@ -232,4 +232,10 @@ config.castTypes = false;
  */
 config.maxRulesByCorr = 20;
 
+/**
+ * Check nonSignal rules using API ngsiv2 or just access to mongo
+ * @type {Boolean}
+ */
+config.nonSignalByAPI = false;
+
 module.exports = config;

lib/models/entitiesStore.js

@@ -28,15 +28,89 @@ var async = require('async'),
     entitiesCollectionName = require('../../config').orionDb.collection,
     myutils = require('../myutils'),
     constants = require('../constants'),
-    logger = require('logops');
+    logger = require('logops'),
+    NGSI = require('ngsijs');
 
 function orionServiceDb(service) {
     return appContext.OrionDb(config.orionDb.prefix + '-' + service);
 }
 
-function findSilentEntities(service, subservice, ruleData, func, callback) {
+function findSilentEntitiesByAPI(service, subservice, ruleData, func, callback) {
+    var context = { op: 'checkNoSignal.findSilentEntitiesByAPI', comp: constants.COMPONENT_NAME };
+
+    var options = {
+        service: service,
+        servicepath: subservice
+    };
+    options.headers = {};
+    // if (token !== null) {
+    //     options.headers[constants.AUTH_HEADER] = token;
+    // }
+    // Add correlator
+    var domain = process.domain;
+    if (domain && domain.context) {
+        options.headers[constants.CORRELATOR_HEADER] = domain.context.corr;
+        // Add other headers
+        if (domain.context.srv && options.headers[constants.SERVICE_HEADER] === undefined) {
+            options.headers[constants.SERVICE_HEADER] = domain.context.srv;
+        }
+        if (domain.context.subsrv && options.headers[constants.SUBSERVICE_HEADER] === undefined) {
+            options.headers[constants.SUBSERVICE_HEADER] = domain.context.subsrv;
+        }
+        if (domain.context.from && options.headers[constants.REALIP_HEADER] === undefined) {
+            options.headers[constants.REALIP_HEADER] = domain.context.from;
+        }
+    }
+    var connection = new NGSI.Connection(config.orion.URL, options);
+
+    var filter = {
+        service: service,
+        servicepath: subservice,
+        type: ruleData.type,
+        mq: ruleData.attribute + '.dateModified<' + (Date.now() / 1000 - ruleData.reportInterval).toString()
+    };
+    if (ruleData.id) {
+        filter.id = ruleData.id;
+    } else if (ruleData.idRegexp) {
+        filter.idPattern = ruleData.idRegexp;
+    }
+    logger.debug(
+        context,
+        'find silent entities by API ngsi using options %j and filter %j and rule %j',
+        options,
+        filter,
+        ruleData
+    );
+    // https://ficodes.github.io/ngsijs/stable/NGSI.Connection.html#.%22v2.listEntities%22__anchor
+    connection.v2.listEntities(filter).then(
+        (response) => {
+            // Entities retrieved successfully
+            // response.correlator transaction id associated with the server response
+            // response.limit contains the used page size
+            // response.results is an array with the retrieved entities
+            // response.offset contains the offset used in the request
+            var count = 0;
+            response.results.forEach((entity) => {
+                logger.debug(context, 'silent entity %j', entity);
+                func(entity);
+                count++;
+            });
+            logger.debug(context, 'findSilentEntities %s', myutils.firstChars(response.results));
+            callback(null, response.results);
+        },
+        (error) => {
+            // Error retrieving entities
+            // If the error was reported by Orion, error.correlator will be
+            // filled with the associated transaction id
+            logger.warn('error v2.listEntities: %j trying list entities using filter %j', error, filter);
+            callback(error, null);
+        }
+    );
+}
+
+function findSilentEntitiesByMongo(service, subservice, ruleData, func, callback) {
     var db,
-        context = { op: 'checkNoSignal', comp: constants.COMPONENT_NAME },
+        context = { op: 'checkNoSignal.findSilentEntitiesByMongo', comp: constants.COMPONENT_NAME },
         criterion = {};
 
     db = orionServiceDb(service);
@@ -88,6 +162,14 @@ function findSilentEntities(service, subservice, ruleData, func, callback) {
     );
 }
 
-module.exports = {
-    FindSilentEntities: findSilentEntities
-};
+function findSilentEntities(service, subservice, ruleData, func, callback) {
+    if (!config.nonSignalByAPI) {
+        return findSilentEntitiesByMongo(service, subservice, ruleData, func, callback);
+    } else {
+        return findSilentEntitiesByAPI(service, subservice, ruleData, func, callback);
+    }
+}
+
+module.exports.FindSilentEntities = findSilentEntities;
+module.exports.findSilentEntitiesByAPI = findSilentEntitiesByAPI;
+module.exports.findSilentEntitiesByMongo = findSilentEntitiesByMongo;

lib/models/noSignal.js

@@ -63,36 +63,63 @@ function alertFunc(nsLineRule, entity) {
         d.exit();
     });
     d.run(function() {
+        logger.debug(context, 'alertfunc nsLineRule %j  entity %j ', nsLineRule, entity);
         // We duplicate info in event and event.ev for VR and non-VR action parameters
         var event = {
             service: nsLineRule[SERVICE],
             subservice: nsLineRule[SUBSERVICE],
             ruleName: nsLineRule[NAME],
             reportInterval: nsLineRule[REPORT_INTERVAL],
-            id: entity._id.id,
-            type: entity._id.type,
             internalCurrentTime: new Date().toISOString()
         };
-
-        // Search for modDate of the entity's attribute
-        // and copy every attribute (if not in event yet)
-        // for use in action template
-        Object.keys(entity.attrs).forEach(function(attrName) {
-            if (attrName === nsLineRule[ATTRIBUTE]) {
+        if (!config.nonSignalByAPI) {
+            // entity is really a entity doc obtained from mongo
+            event.id = entity._id.id;
+            event.type = entity._id.type;
+            logger.debug(context, 'alertfunc event %j ', event);
+            // Search for modDate of the entity's attribute
+            // and copy every attribute (if not in event yet)
+            // for use in action template
+            Object.keys(entity.attrs).forEach(function(attrName) {
+                if (attrName === nsLineRule[ATTRIBUTE]) {
+                    try {
+                        lastTime = new Date(entity.attrs[attrName].modDate * 1000).toISOString();
+                    } catch (ex) {
+                        myutils.logErrorIf(ex, 'run ', d.context);
+                    }
+                }
+                if (event[attrName] === undefined) {
+                    if (entity.attrs[attrName].type === 'DateTime') {
+                        event[attrName] = new Date(entity.attrs[attrName].value * 1000).toISOString();
+                    } else {
+                        event[attrName] = entity.attrs[attrName].value;
+                    }
+                }
+            });
+        } else {
+            // entity is and NGSI object
+            event.id = entity.id;
+            event.type = entity.type;
+            logger.debug(context, 'alertfunc event %j ', event);
+            // Search for modDate of the entity's attribute
+            // and copy every attribute (if not in event yet)
+            // for use in action template
+            const attrName = nsLineRule[ATTRIBUTE];
+            if (entity[attrName]) {
                 try {
-                    lastTime = new Date(entity.attrs[attrName].modDate * 1000).toISOString();
+                    lastTime = entity[attrName].metadata.TimeInstant.value;
                 } catch (ex) {
                     myutils.logErrorIf(ex, 'run ', d.context);
                 }
-            }
-            if (event[attrName] === undefined) {
-                if (entity.attrs[attrName].type === 'DateTime') {
-                    event[attrName] = new Date(entity.attrs[attrName].value * 1000).toISOString();
-                } else {
-                    event[attrName] = entity.attrs[attrName].value;
+                if (event[attrName] === undefined) {
+                    if (entity[attrName].type === 'DateTime') {
+                        event[attrName] = entity[attrName].metadata.TimeInstant.value;
+                    } else {
+                        event[attrName] = entity[attrName].value;
+                    }
                 }
             }
-        });
+        }
 
         logger.debug(context, 'lastTime could be ', lastTime);
         if (lastTime !== undefined && lastTime !== null) {
@@ -119,6 +146,7 @@ function checkNoSignal(period) {
     currentContext.srv = 'n/a';
     currentContext.subsrv = 'n/a';
     logger.debug(currentContext, 'Executing no-signal handler for period of %d (%d rules)', period, list.length);
+
     list.forEach(function(nsrule) {
         currentContext.srv = nsrule[SERVICE];
         currentContext.subsrv = nsrule[SUBSERVICE];
@@ -192,6 +220,7 @@ function addNSRule(service, subservice, name, nsr) {
         );
         intervalAsNum = MIN_INTERVAL_MS;
     }
+
     arrayRule = nsr2arr(service, subservice, name, nsr);
     nsRulesByInterval[nsr.checkInterval] = nsRulesByInterval[nsr.checkInterval] || [];
     nsRulesByInterval[nsr.checkInterval].forEach(function(element, index, array) {
@@ -206,6 +235,7 @@ function addNSRule(service, subservice, name, nsr) {
         logger.debug(context, util.format('Adding no-signal rule (%s, %s, %s)', service, subservice, name));
     }
     if (!checkers.hasOwnProperty(nsr.checkInterval)) {
+        logger.info(context, util.format('no-signal rule (%s, %s, %s)', service, subservice, name));
         checkers[nsr.checkInterval] = setInterval(checkNoSignal, intervalAsNum, nsr.checkInterval);
         checkers[nsr.checkInterval].unref();
     }

test/unit/entitiesStore_utest.js

@@ -0,0 +1,126 @@
+/*
+ * Copyright 2015 Telefonica Investigación y Desarrollo, S.A.U
+ *
+ * This file is part of perseo-fe
+ *
+ * perseo-fe is free software: you can redistribute it and/or
+ * modify it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the License,
+ * or (at your option) any later version.
+ *
+ * perseo-fe is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
+ * See the GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public
+ * License along with perseo-fe.
+ * If not, see http://www.gnu.org/licenses/.
+ *
+ * For those usages not covered by the GNU Affero General Public License
+ * please contact with iot_support at tid dot es
+ *
+ * Created by: Carlos Blanco - Future Internet Consulting and Development Solutions (FICODES)
+ */
+
+'use strict';
+
+var should = require('should');
+var rewire = require('rewire');
+var entitiesStore = rewire('../../lib/models/entitiesStore.js');
+var chai = require('chai');
+var sinon = require('sinon');
+var sinonChai = require('sinon-chai');
+var expect = chai.expect;
+var config = require('../../config.js');
+var NGSI = require('ngsijs');
+chai.Should();
+chai.use(sinonChai);
+
+describe('entitiesStore', function() {
+    describe('#findSilentEntitiesByAPI', function() {
+        var connectionMock;
+        var listEntitiesMock;
+
+        beforeEach(function() {
+            connectionMock = sinon.stub(NGSI, 'Connection');
+            listEntitiesMock = sinon.stub();
+
+            // Mock the Connection function
+            connectionMock.returns({ v2: { listEntities: listEntitiesMock } });
+        });
+
+        afterEach(function() {
+            // Restore the original function after each test
+            connectionMock.restore();
+        });
+
+        var ruleData = {
+                name: 'NSR2',
+                action: {
+                    type: 'update',
+                    parameters: {
+                        id: 'alarma:${id}',
+                        type: 'Alarm',
+                        attributes: [
+                            {
+                                name: 'msg',
+                                value: 'El status de ${id} es ${status}'
+                            }
+                        ]
+                    }
+                },
+                subservice: '/',
+                service: 'unknownt',
+                nosignal: {
+                    checkInterval: '1',
+                    attribute: 'temperature',
+                    reportInterval: '5',
+                    id: 'thing:disp1',
+                    idRegexp: null,
+                    type: 'thing'
+                }
+            },
+            func = 'sinon.stub()',
+            callback = function(e, request) {
+                should.exist(request);
+                should.not.exist(e);
+                should.equal(request.httpCode, 200);
+            };
+
+        it('By default should call findSilentEntitiesByMongo', function() {
+            var findSilentEntitiesByMongoSpy = sinon.spy();
+            entitiesStore.__set__('findSilentEntitiesByMongo', findSilentEntitiesByMongoSpy);
+            entitiesStore.FindSilentEntities(ruleData.service, ruleData.subservice, ruleData, func, callback);
+            sinon.assert.calledOnce(findSilentEntitiesByMongoSpy);
+        });
+
+        it('If default settings are changed FindSilentEntitiesByAPI should be called', function() {
+            config.nonSignalByAPI = true;
+            var findSilentEntitiesByAPISpy = sinon.spy();
+            entitiesStore.__set__('findSilentEntitiesByAPI', findSilentEntitiesByAPISpy);
+            entitiesStore.FindSilentEntities();
+            sinon.assert.calledOnce(findSilentEntitiesByAPISpy);
+        });
+
+        it('should return silent entities', async function() {
+            var funcM = sinon.spy(),
+                callbackM = sinon.spy();
+
+            // Mock the listEntities function to resolve with a test response
+            listEntitiesMock.returns(Promise.resolve({ results: [] }));
+
+            await entitiesStore.findSilentEntitiesByAPI(
+                ruleData.service,
+                ruleData.subservice,
+                ruleData,
+                funcM,
+                callbackM
+            );
+
+            expect(listEntitiesMock.calledOnce).to.be.true;
+            expect(funcM.callCount).to.equal(0);
+            expect(callbackM.calledOnceWith(null, [])).to.be.true;
+        });
+    });
+});