14240036685

Committed 03 Apr 2025 09:48AM UTC coverage: 78.211% (-0.1%) from 78.308%

Build # 14240036685

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #827 from telefonicaid/task/add_silent_entities_time_limit

Task/add silent entities check time limit

Run Details

741 of 1072 branches covered (69.12%)

Branch coverage included in aggregate %.

14 of 16 new or added lines in 3 files covered. (87.5%)

1 existing line in 1 file now uncovered.

2170 of 2650 relevant lines covered (81.89%)

70.41 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

60.77

/lib/models/entitiesStore.js

/*
 * 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::[contacto@tid.es]
 */
'use strict';

var async = require('async'),
    appContext = require('../appContext'),
    config = require('../../config'),
    entitiesCollectionName = require('../../config').orionDb.collection,
    myutils = require('../myutils'),
    constants = require('../constants'),
    logger = require('logops'),
    ngsi = require('ngsijs'),
    context = { op: 'entitiesStore', comp: constants.COMPONENT_NAME };

function orionServiceDb(service) {
    return appContext.OrionDb(config.orionDb.prefix + '-' + service);
}

function createFilter(ruleData, service, subservice, limit, offset) {
    var maxTimeDetection = ruleData.maxTimeDetection ? ruleData.maxTimeDetection : config.nonSignalMaxTimeDetection;
    var currentTime = Date.now() / 1000;
    var filter = {
        service: service,
        servicepath: subservice,
        type: ruleData.type,
        mq:
            ruleData.attribute +
            '.dateModified<' +
            (currentTime - ruleData.reportInterval).toString() +
            ';' +
            'temperature.dateModified>' +
            (currentTime / 1000 - maxTimeDetection).toString(),
        // TO DO: Use ruleData.maxTimeDetectionAttr and ruleData.reportIntervalAttr in query if possible (#824)
        // https://github.com/telefonicaid/fiware-orion/blob/master/doc/manuals/orion-api.md#simple-query-language
        // https://ficodes.github.io/ngsijs/stable/NGSI.Connection.html#.%22v2.listEntities%22
        limit: limit,
        offset: offset
    };
    if (ruleData.id) {
        filter.id = ruleData.id;
    } else if (ruleData.idRegexp) {
        filter.idPattern = ruleData.idRegexp;
    }
    return filter;
}

function createConnection(service, subservice) {
    var options = {
        service: service,
        servicepath: subservice
    };
    options.headers = {};
    // 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;
        }
    }

    return new ngsi.Connection(config.orion.URL, options);
}

function findSilentEntitiesByAPIWithPagination(connection, filter, alterFunc, accumulatedResults, callback) {
    connection.v2.listEntities(filter).then(
        (response) => {
            if (response.results) {
                response.results.forEach((entity) => {
                    logger.debug(context, 'silent entity %j', entity);
                    alterFunc(entity);
                    accumulatedResults.push(entity);
                });
            }
            logger.debug(context, 'findSilentEntities %s', myutils.firstChars(response.results));
            if (response.count > filter.limit + filter.offset) {
                filter.offset += filter.limit;
                findSilentEntitiesByAPIWithPagination(connection, filter, alterFunc, accumulatedResults, callback);
            } else {
                // Pass the count of entities to the callback
                callback(null, accumulatedResults, accumulatedResults.length);
            }
        },
        (error) => {
            logger.warn(context, ' error v2.listEntities: %j trying list entities using filter %j', error, filter);
            callback(error, null, null);
        }
    );
}

function findSilentEntitiesByAPI(service, subservice, ruleData, alterFunc, callback) {
    var limit = 20;
    var offset = 0;
    var connection = createConnection(service, subservice);
    var filter = createFilter(ruleData, service, subservice, limit, offset);

    logger.info(
        context,
        ' find silent entities by API ngsi using options %j and filter %j and rule %j',
        connection,
        filter,
        ruleData
    );

    var accumulatedResults = [];

    findSilentEntitiesByAPIWithPagination(connection, filter, alterFunc, accumulatedResults, callback);
}

function findSilentEntitiesByMongo(service, subservice, ruleData, alterFunc, callback) {
    var db,
        criterion = {};

    db = orionServiceDb(service);
    criterion['_id.servicePath'] = subservice;
    if (ruleData.id) {
        criterion['_id.id'] = ruleData.id;
    } else if (ruleData.idRegexp) {
        try {
            criterion['_id.id'] = new RegExp(ruleData.idRegexp);
        } catch (e) {
            return callback(e, null);
        }
    }
    if (ruleData.type) {
        criterion['_id.type'] = ruleData.type;
    }
    // Variable to store the count of entities
    var entityCount = 0;

    async.waterfall(
        [
            db.collection.bind(db, entitiesCollectionName, { strict: false }),
            function(col, cb) {
                var pipeline = [
                    {
                        $match: criterion
                    },
                    {
                        $addFields: {
                            currentTime: { $divide: [Date.now(), 1000] },
                            maxTimeDetection: {
                                $convert: {
                                    input: {
                                        $ifNull: [
                                            '$attrs.' + ruleData.maxTimeDetectionAttr + '.value',
                                            ruleData.maxTimeDetection !== undefined
                                                ? ruleData.maxTimeDetection // jshint ignore: line
                                                : config.nonSignalMaxTimeDetection
                                        ]
                                    },
                                    to: 'double',
                                    onError: config.nonSignalMaxTimeDetection,
                                    onNull: config.nonSignalMaxTimeDetection
                                }
                            },
                            reportInterval: {
                                $convert: {
                                    input: {
                                        $ifNull: [
                                            '$attrs.' + ruleData.reportIntervalAttr + '.value',
                                            ruleData.reportInterval
                                        ]
                                    },
                                    to: 'double',
                                    onError: ruleData.reportInterval,
                                    onNull: ruleData.reportInterval
                                }
                            }
                        }
                    },
                    {
                        $match: {
                            $expr: {
                                $and: [
                                    {
                                        $lt: [
                                            '$attrs.' + ruleData.attribute + '.modDate',
                                            { $subtract: ['$currentTime', '$reportInterval'] }
                                        ]
                                    },
                                    {
                                        $gt: [
                                            '$attrs.' + ruleData.attribute + '.modDate',
                                            { $subtract: ['$currentTime', '$maxTimeDetection'] }
                                        ]
                                    },
                                    {
                                        $gt: ['$attrs.' + ruleData.attribute + '.modDate', 0]
                                    } // exclude invalid dates
                                ]
                            }
                        }
                    }
                ];
                logger.debug(context, 'findSilentEntities service %s pipeline: %j ', service, pipeline);
                col.aggregate(pipeline).toArray(function(err, results) {
                    if (err) {
                        return cb(err, null);
                    }
                    results.forEach(function(one) {
                        logger.debug(context, 'silent entity %j', one._id);
                        alterFunc(one);
                        // Increment the count of entities
                        entityCount++;
                    });
                    cb(null, 'silent ones count ' + entityCount);
                });
            }
        ],
        function(err, result) {
            logger.debug(context, 'findSilentEntities %s', myutils.firstChars(result));
            return callback(err, result, entityCount);
        }
    );
}

function findSilentEntities(service, subservice, ruleData, alterFunc, callback) {
    var hrstart = process.hrtime();
    var method = !config.nonSignalByAPI ? 'findSilentEntitiesByMongo' : 'findSilentEntitiesByAPI';

    var timedCallback = function(err, result, entityCount) {
        var hrend = process.hrtime(hrstart);

        logger.info(context, ' %s has found %d entities in (hr): %d ms', method, entityCount, hrend[1] / 1000000);

        callback(err, result, entityCount);
    };
    if (!config.nonSignalByAPI) {
        return findSilentEntitiesByMongo(service, subservice, ruleData, alterFunc, timedCallback);
    } else {
        return findSilentEntitiesByAPI(service, subservice, ruleData, alterFunc, timedCallback);
    }
}

module.exports.FindSilentEntities = findSilentEntities;
module.exports.findSilentEntitiesByAPI = findSilentEntitiesByAPI;
module.exports.findSilentEntitiesByMongo = findSilentEntitiesByMongo;
module.exports.findSilentEntitiesByAPIWithPagination = findSilentEntitiesByAPIWithPagination;
module.exports.createConnection = createConnection;
module.exports.createFilter = createFilter;

1	/*
2	* Copyright 2015 Telefonica Investigación y Desarrollo, S.A.U
3	*
4	* This file is part of perseo-fe
5	*
6	* perseo-fe is free software: you can redistribute it and/or
7	* modify it under the terms of the GNU Affero General Public License as
8	* published by the Free Software Foundation, either version 3 of the License,
9	* or (at your option) any later version.
10	*
11	* perseo-fe is distributed in the hope that it will be useful,
12	* but WITHOUT ANY WARRANTY; without even the implied warranty of
13	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
14	* See the GNU Affero General Public License for more details.
15	*
16	* You should have received a copy of the GNU Affero General Public
17	* License along with perseo-fe.
18	* If not, see http://www.gnu.org/licenses/.
19	*
20	* For those usages not covered by the GNU Affero General Public License
21	* please contact with::[contacto@tid.es]
22	*/
23	'use strict';
24
25	var async = require('async'),	2✔
26	appContext = require('../appContext'),	2✔
27	config = require('../../config'),	2✔
28	entitiesCollectionName = require('../../config').orionDb.collection,	2✔
29	myutils = require('../myutils'),	2✔
30	constants = require('../constants'),	2✔
31	logger = require('logops'),	2✔
32	ngsi = require('ngsijs'),	2✔
33	context = { op: 'entitiesStore', comp: constants.COMPONENT_NAME };	2✔
34
35	function orionServiceDb(service) {
36	return appContext.OrionDb(config.orionDb.prefix + '-' + service);	1✔
37	}
38
39	function createFilter(ruleData, service, subservice, limit, offset) {
NEW 40	var maxTimeDetection = ruleData.maxTimeDetection ? ruleData.maxTimeDetection : config.nonSignalMaxTimeDetection;	×
NEW 41	var currentTime = Date.now() / 1000;	×
UNCOV 42	var filter = {	×
43	service: service,
44	servicepath: subservice,
45	type: ruleData.type,
46	mq:
47	ruleData.attribute +
48	'.dateModified<' +
49	(currentTime - ruleData.reportInterval).toString() +
50	';' +
51	'temperature.dateModified>' +
52	(currentTime / 1000 - maxTimeDetection).toString(),
53	// TO DO: Use ruleData.maxTimeDetectionAttr and ruleData.reportIntervalAttr in query if possible (#824)
54	// https://github.com/telefonicaid/fiware-orion/blob/master/doc/manuals/orion-api.md#simple-query-language
55	// https://ficodes.github.io/ngsijs/stable/NGSI.Connection.html#.%22v2.listEntities%22
56	limit: limit,
57	offset: offset
58	};
59	if (ruleData.id) {	×
60	filter.id = ruleData.id;	×
61	} else if (ruleData.idRegexp) {	×
62	filter.idPattern = ruleData.idRegexp;	×
63	}
64	return filter;	×
65	}
66
67	function createConnection(service, subservice) {
68	var options = {	×
69	service: service,
70	servicepath: subservice
71	};
72	options.headers = {};	×
73	// Add correlator
74	var domain = process.domain;	×
75	if (domain && domain.context) {	×
76	options.headers[constants.CORRELATOR_HEADER] = domain.context.corr;	×
77	// Add other headers
78	if (domain.context.srv && options.headers[constants.SERVICE_HEADER] === undefined) {	×
79	options.headers[constants.SERVICE_HEADER] = domain.context.srv;	×
80	}
81	if (domain.context.subsrv && options.headers[constants.SUBSERVICE_HEADER] === undefined) {	×
82	options.headers[constants.SUBSERVICE_HEADER] = domain.context.subsrv;	×
83	}
84	if (domain.context.from && options.headers[constants.REALIP_HEADER] === undefined) {	×
85	options.headers[constants.REALIP_HEADER] = domain.context.from;	×
86	}
87	}
88
89	return new ngsi.Connection(config.orion.URL, options);	×
90	}
91
92	function findSilentEntitiesByAPIWithPagination(connection, filter, alterFunc, accumulatedResults, callback) {
93	connection.v2.listEntities(filter).then(	2✔
94	(response) => {
95	if (response.results) {	1!
96	response.results.forEach((entity) => {	1✔
97	logger.debug(context, 'silent entity %j', entity);	2✔
98	alterFunc(entity);	2✔
99	accumulatedResults.push(entity);	2✔
100	});
101	}
102	logger.debug(context, 'findSilentEntities %s', myutils.firstChars(response.results));	1✔
103	if (response.count > filter.limit + filter.offset) {	1!
104	filter.offset += filter.limit;	1✔
105	findSilentEntitiesByAPIWithPagination(connection, filter, alterFunc, accumulatedResults, callback);	1✔
106	} else {
107	// Pass the count of entities to the callback
108	callback(null, accumulatedResults, accumulatedResults.length);	×
109	}
110	},
111	(error) => {
112	logger.warn(context, ' error v2.listEntities: %j trying list entities using filter %j', error, filter);	1✔
113	callback(error, null, null);	1✔
114	}
115	);
116	}
117
118	function findSilentEntitiesByAPI(service, subservice, ruleData, alterFunc, callback) {
119	var limit = 20;	1✔
120	var offset = 0;	1✔
121	var connection = createConnection(service, subservice);	1✔
122	var filter = createFilter(ruleData, service, subservice, limit, offset);	1✔
123
124	logger.info(	1✔
125	context,
126	' find silent entities by API ngsi using options %j and filter %j and rule %j',
127	connection,
128	filter,
129	ruleData
130	);
131
132	var accumulatedResults = [];	1✔
133
134	findSilentEntitiesByAPIWithPagination(connection, filter, alterFunc, accumulatedResults, callback);	1✔
135	}
136
137	function findSilentEntitiesByMongo(service, subservice, ruleData, alterFunc, callback) {
138	var db,
139	criterion = {};	3✔
140
141	db = orionServiceDb(service);	3✔
142	criterion['_id.servicePath'] = subservice;	3✔
143	if (ruleData.id) {	3!
144	criterion['_id.id'] = ruleData.id;	×
145	} else if (ruleData.idRegexp) {	3✔
146	try {	1✔
147	criterion['_id.id'] = new RegExp(ruleData.idRegexp);	1✔
148	} catch (e) {
149	return callback(e, null);	×
150	}
151	}
152	if (ruleData.type) {	3!
153	criterion['_id.type'] = ruleData.type;	×
154	}
155	// Variable to store the count of entities
156	var entityCount = 0;	3✔
157
158	async.waterfall(	3✔
159	[
160	db.collection.bind(db, entitiesCollectionName, { strict: false }),
161	function(col, cb) {
162	var pipeline = [	3✔
163	{
164	$match: criterion
165	},
166	{
167	$addFields: {
168	currentTime: { $divide: [Date.now(), 1000] },
169	maxTimeDetection: {
170	$convert: {
171	input: {
172	$ifNull: [
173	'$attrs.' + ruleData.maxTimeDetectionAttr + '.value',
174	ruleData.maxTimeDetection !== undefined	3!
175	? ruleData.maxTimeDetection // jshint ignore: line
176	: config.nonSignalMaxTimeDetection
177	]
178	},
179	to: 'double',
180	onError: config.nonSignalMaxTimeDetection,
181	onNull: config.nonSignalMaxTimeDetection
182	}
183	},
184	reportInterval: {
185	$convert: {
186	input: {
187	$ifNull: [
188	'$attrs.' + ruleData.reportIntervalAttr + '.value',
189	ruleData.reportInterval
190	]
191	},
192	to: 'double',
193	onError: ruleData.reportInterval,
194	onNull: ruleData.reportInterval
195	}
196	}
197	}
198	},
199	{
200	$match: {
201	$expr: {
202	$and: [
203	{
204	$lt: [
205	'$attrs.' + ruleData.attribute + '.modDate',
206	{ $subtract: ['$currentTime', '$reportInterval'] }
207	]
208	},
209	{
210	$gt: [
211	'$attrs.' + ruleData.attribute + '.modDate',
212	{ $subtract: ['$currentTime', '$maxTimeDetection'] }
213	]
214	},
215	{
216	$gt: ['$attrs.' + ruleData.attribute + '.modDate', 0]
217	} // exclude invalid dates
218	]
219	}
220	}
221	}
222	];
223	logger.debug(context, 'findSilentEntities service %s pipeline: %j ', service, pipeline);	3✔
224	col.aggregate(pipeline).toArray(function(err, results) {	3✔
225	if (err) {	2✔
226	return cb(err, null);	1✔
227	}
228	results.forEach(function(one) {	1✔
229	logger.debug(context, 'silent entity %j', one._id);	2✔
230	alterFunc(one);	2✔
231	// Increment the count of entities
232	entityCount++;	2✔
233	});
234	cb(null, 'silent ones count ' + entityCount);	1✔
235	});
236	}
237	],
238	function(err, result) {
239	logger.debug(context, 'findSilentEntities %s', myutils.firstChars(result));	2✔
240	return callback(err, result, entityCount);	2✔
241	}
242	);
243	}
244
245	function findSilentEntities(service, subservice, ruleData, alterFunc, callback) {
246	var hrstart = process.hrtime();	3✔
247	var method = !config.nonSignalByAPI ? 'findSilentEntitiesByMongo' : 'findSilentEntitiesByAPI';	3✔
248
249	var timedCallback = function(err, result, entityCount) {	3✔
250	var hrend = process.hrtime(hrstart);	1✔
251
252	logger.info(context, ' %s has found %d entities in (hr): %d ms', method, entityCount, hrend[1] / 1000000);	1✔
253
254	callback(err, result, entityCount);	1✔
255	};
256	if (!config.nonSignalByAPI) {	3✔
257	return findSilentEntitiesByMongo(service, subservice, ruleData, alterFunc, timedCallback);	2✔
258	} else {
259	return findSilentEntitiesByAPI(service, subservice, ruleData, alterFunc, timedCallback);	1✔
260	}
261	}
262
263	module.exports.FindSilentEntities = findSilentEntities;	2✔
264	module.exports.findSilentEntitiesByAPI = findSilentEntitiesByAPI;	2✔
265	module.exports.findSilentEntitiesByMongo = findSilentEntitiesByMongo;	2✔
266	module.exports.findSilentEntitiesByAPIWithPagination = findSilentEntitiesByAPIWithPagination;	2✔
267	module.exports.createConnection = createConnection;	2✔
268	module.exports.createFilter = createFilter;	2✔

telefonicaid / perseo-fe / 14240036685

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous