Skip to content

Commit d3e8983

Browse files
ar2rsawseenar2rsawseen
committed
Merge branch 'master' into flex
2 parents d1ead03 + 1a8efe1 commit d3e8983

File tree

40 files changed

+1200
-880
lines changed

40 files changed

+1200
-880
lines changed

‎CHANGELOG.md

+38
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,41 @@
1+
## Version 24.05.11
2+
Fixes:
3+
- [cache] Use a cursor without timeout
4+
5+
## Version 24.05.10
6+
Fixes:
7+
- [alerts] Alerts improvements
8+
- [core] Various fixes for frontend to support running countly from subdirectory
9+
- [logs] Show collected problems on logger
10+
11+
Enterprise fixes:
12+
- [data-manager] Fixes n UI to allow events starting with "/"
13+
- [flows] Fixes for flows step generation
14+
- [surveys] Other is not allowed as a valid answer for required questions
15+
16+
## Version 24.05.9
17+
Fixes:
18+
- [crashes] Fix crashes template paths and add data check
19+
20+
Enterprise fixes:
21+
- [license] Fix for License warning appears is Flex server
22+
- [drill] Fix typo in view regeneration api response
23+
24+
## Version 24.05.8
25+
Fixes:
26+
- [scripts] script to check settings limits
27+
- [crashes] Update custom field cleanup script with streaming
28+
- [core] Skip data masking on internal appUsers data refetching query
29+
- [core][bugfix] Correct validation on "is any plugin masked"
30+
- [scripts] Delete custom events scripts
31+
- [dashboards] Show edit option also for invalid widgets
32+
- [core] batcher stats added
33+
- [dependencies] Bump xlsx-write-stream from 1.0.2 to 1.0.3
34+
35+
Enterprise fixes:
36+
- [data-manager] changes to allow skipping query rewriting using passed property
37+
- [os] fix for changing default repository of CentOS 8 to vault because OS reached EOL
38+
139
## Version 24.05.7
240
Fixes:
341
- [countly-request]Fix countly-request get and post methods

‎api/parts/data/batcher.js

+36-3
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,24 @@ const plugins = require('../../../plugins/pluginManager.js');
44
const log = require('../../utils/log.js')("batcher");
55
const common = require('../../utils/common.js');
66

7-
7+
var batcherStats = {
8+
key: 'BATCHER_STATS',
9+
pid: process.pid,
10+
insert_queued: 0,
11+
insert_processing: 0,
12+
insert_errored_fallback: 0,
13+
insert_errored_no_fallback: 0,
14+
insert_errored_no_fallback_last_error: "",
15+
update_queued: 0,
16+
update_processing: 0,
17+
update_errored_fallback: 0,
18+
update_errored_no_fallback: 0,
19+
update_errored_no_fallback_last_error: "",
20+
};
21+
22+
setInterval(function() {
23+
log.i('%j', batcherStats);
24+
}, 10000);
825
/**
926
* Class for batching database insert operations
1027
* @example
@@ -56,13 +73,16 @@ class InsertBatcher {
5673
if (this.data[db][collection].length) {
5774
var docs = this.data[db][collection];
5875
this.data[db][collection] = [];
76+
batcherStats.insert_queued -= docs.length;
77+
batcherStats.insert_processing += docs.length;
5978
try {
6079
await new Promise((resolve, reject) => {
6180
this.dbs[db].collection(collection).insertMany(docs, {ordered: false, ignore_errors: [11000]}, function(err, res) {
6281
if (err) {
6382
reject(err);
6483
return;
6584
}
85+
batcherStats.insert_processing -= docs.length;
6686
resolve(res);
6787
});
6888
});
@@ -75,7 +95,10 @@ class InsertBatcher {
7595
//trying to rollback operations to try again on next iteration
7696
if (ex.writeErrors && ex.writeErrors.length) {
7797
for (let i = 0; i < ex.writeErrors.length; i++) {
98+
batcherStats.insert_processing--;
7899
if (no_fallback_errors.indexOf(ex.writeErrors[i].code) !== -1) {
100+
batcherStats.insert_errored_no_fallback++;
101+
batcherStats.insert_errored_no_fallback_last_error = ex.writeErrors[i].errmsg;
79102
//dispatch failure
80103
if (notify_fallback_errors.indexOf(ex.writeErrors[i].code) !== -1) {
81104
var index0 = ex.writeErrors[i].index;
@@ -85,6 +108,7 @@ class InsertBatcher {
85108
//we could record in diagnostic data
86109
continue;
87110
}
111+
batcherStats.insert_errored_fallback++;
88112
let index = ex.writeErrors[i].index;
89113
if (docs[index]) {
90114
this.data[db][collection].push(docs[index]);
@@ -136,8 +160,10 @@ class InsertBatcher {
136160
for (let i = 0; i < doc.length; i++) {
137161
this.data[db][collection].push(doc[i]);
138162
}
163+
batcherStats.insert_queued += doc.length;
139164
}
140165
else {
166+
batcherStats.insert_queued++;
141167
this.data[db][collection].push(doc);
142168
}
143169
if (!this.process) {
@@ -212,13 +238,16 @@ class WriteBatcher {
212238
}
213239
}
214240
this.data[db][collection] = {};
241+
batcherStats.update_queued -= queries.length;
242+
batcherStats.update_processing += queries.length;
215243
try {
216244
await new Promise((resolve, reject) => {
217245
this.dbs[db].collection(collection).bulkWrite(queries, {ordered: false, ignore_errors: [11000]}, function(err, res) {
218246
if (err) {
219247
reject(err);
220248
return;
221249
}
250+
batcherStats.update_processing -= queries.length;
222251
resolve(res);
223252
});
224253
});
@@ -231,8 +260,10 @@ class WriteBatcher {
231260
//trying to rollback operations to try again on next iteration
232261
if (ex.writeErrors && ex.writeErrors.length) {
233262
for (let i = 0; i < ex.writeErrors.length; i++) {
234-
263+
batcherStats.update_processing--;
235264
if (no_fallback_errors.indexOf(ex.writeErrors[i].code) !== -1) {
265+
batcherStats.update_errored_no_fallback++;
266+
batcherStats.update_errored_no_fallback_last_error = ex.writeErrors[i].errmsg;
236267
//dispatch failure
237268
if (notify_errors.indexOf(ex.writeErrors[i].code) !== -1) {
238269
var index0 = ex.writeErrors[i].index;
@@ -242,6 +273,7 @@ class WriteBatcher {
242273
//we could record in diagnostic data
243274
continue;
244275
}
276+
batcherStats.update_errored_fallback++;
245277
let index = ex.writeErrors[i].index;
246278
if (queries[index]) {
247279
//if we don't have anything for this document yet just use query
@@ -316,6 +348,7 @@ class WriteBatcher {
316348
}
317349
if (!this.data[db][collection][id]) {
318350
this.data[db][collection][id] = {id: id, value: operation};
351+
batcherStats.update_queued++;
319352
}
320353
else {
321354
this.data[db][collection][id].value = common.mergeQuery(this.data[db][collection][id].value, operation);
@@ -653,4 +686,4 @@ function getId() {
653686
return crypto.randomBytes(16).toString("hex");
654687
}
655688

656-
module.exports = {WriteBatcher, ReadBatcher, InsertBatcher};
689+
module.exports = {WriteBatcher, ReadBatcher, InsertBatcher};

‎api/parts/data/cache.js

+1-1
Original file line numberDiff line numberDiff line change
@@ -971,7 +971,7 @@ class StreamedCollection {
971971
let [col, last] = await createCollection(this.db, this.name, 1e7);
972972

973973
this.col = col;
974-
this.stream = col.find({_id: {$gt: last}}, {tailable: true, awaitData: true, numberOfRetries: -1}).stream();
974+
this.stream = col.find({_id: {$gt: last}}, {tailable: true, awaitData: true, noCursorTimeout: true, numberOfRetries: -1}).stream();
975975

976976
this.stream.on('data', doc => {
977977
if (this.inserts.indexOf(doc._id.toString()) !== -1) {

‎api/utils/common.js

+2-1
Original file line numberDiff line numberDiff line change
@@ -2767,7 +2767,8 @@ common.updateAppUser = function(params, update, no_meta, callback) {
27672767
if (callback) {
27682768
common.db.collection('app_users' + params.app_id).findAndModify({'_id': params.app_user_id}, {}, common.clearClashingQueryOperations(update), {
27692769
new: true,
2770-
upsert: true
2770+
upsert: true,
2771+
skipDataMasking: true
27712772
}, function(err, res) {
27722773
if (!err && res && res.value) {
27732774
params.app_user = res.value;

‎bin/scripts/expire-data/countly_single_app_expireDataBatches.js

+98-11
Original file line numberDiff line numberDiff line change
@@ -171,6 +171,77 @@ function prepareIterationList(collections, seconds) {
171171
return listed;
172172

173173
}
174+
175+
function processDrillCollection(collection, seconds, callback) {
176+
var listed = [];
177+
178+
if (start === 0) {
179+
getMinTs(function(err, minTs) {
180+
if (err) {
181+
console.log("ERROR: Could not fetch min ts for collection " + collection.collection);
182+
callback(err);
183+
}
184+
else {
185+
console.log("Min ts for collection " + collection.collection + ": " + minTs);
186+
generateIterationList(minTs);
187+
processCollection();
188+
}
189+
});
190+
}
191+
else {
192+
generateIterationList(start);
193+
processCollection();
194+
}
195+
196+
function getMinTs(cb) {
197+
collection.db.collection(collection.collection).findOne({}, { sort: { ts: 1 }, projection: { ts: 1 } }, function(err, doc) {
198+
if (err) {
199+
return callback(err);
200+
}
201+
if (doc && doc.ts) {
202+
cb(null, doc.ts);
203+
}
204+
else {
205+
cb(null, end);
206+
}
207+
});
208+
}
209+
210+
function generateIterationList(z) {
211+
z = (start === 0 && z) ? z : start;
212+
if (timeSpan === 0 && start === 0) {
213+
listed.push({"collection": collection.collection, "db": collection.db, "start": 0, "end": end, "query": {"ts": {"$lt": end}}});
214+
}
215+
else if (timeSpan === 0) {
216+
listed.push({"collection": collection.collection, "db": collection.db, "start": z, "end": end, "query": {"ts": {"$gte": z, "$lt": end}}});
217+
}
218+
else {
219+
if (seconds) {
220+
z = Math.floor(z / 1000);
221+
for (; z <= Math.floor(end / 1000); z += timeSpan) {
222+
listed.push({"collection": collection.collection, "db": collection.db, "start": z, "end": Math.min(z + timeSpan, end), "seconds": true});
223+
}
224+
}
225+
else {
226+
for (; z <= end; z += timeSpan * 1000) {
227+
listed.push({"collection": collection.collection, "db": collection.db, "start": z, "end": Math.min(z + timeSpan * 1000, end)});
228+
}
229+
}
230+
}
231+
}
232+
233+
function processCollection() {
234+
async.eachLimit(listed, paralelCn, eventIterator, function(err) {
235+
if (err) {
236+
console.log("ERROR: Error while processing drill collection: " + collection.collection);
237+
return callback(err);
238+
}
239+
console.log('Finished processing collection ' + collection.collection);
240+
callback();
241+
});
242+
}
243+
}
244+
174245
function processDrillCollections(db, drill_db, callback) {
175246
if (process && process.drill_events) {
176247
var collections = [];
@@ -192,10 +263,19 @@ function processDrillCollections(db, drill_db, callback) {
192263
collections.push({'db': drill_db, 'collection': "drill_events" + crypto.createHash('sha1').update(eventData.list[i] + APP_ID).digest('hex')});
193264
}
194265
}
195-
var iteratorList = prepareIterationList(collections);
196-
async.eachLimit(iteratorList, paralelCn, eventIterator, function() {
197-
console.log('Drill collections processed');
198-
callback();
266+
267+
async.eachSeries(collections, function(collection, done) {
268+
processDrillCollection(collection, false, function(err) {
269+
if (err) {
270+
console.log("ERROR: Error while processing drill collection: " + collection.collection);
271+
}
272+
done(err);
273+
});
274+
}, function(err) {
275+
if (err) {
276+
console.log("ERROR: Error processing collections.");
277+
}
278+
callback(err);
199279
});
200280
});
201281
}
@@ -224,15 +304,22 @@ Promise.all([plugins.dbConnection("countly"), plugins.dbConnection("countly_dril
224304
}
225305
}
226306
}
227-
var iteratorList = prepareIterationList(processCols, true);
228-
async.eachLimit(iteratorList, paralelCn, eventIterator, function() {
229-
if (errorCn > 0) {
230-
console.log("There were errors. Please recheck logs for those.");
231-
}
232-
console.log('finished');
307+
if (processCols.length === 0) {
308+
console.log("Finished");
233309
db.close();
234310
db_drill.close();
235-
});
311+
}
312+
else {
313+
var iteratorList = prepareIterationList(processCols, true);
314+
async.eachLimit(iteratorList, paralelCn, eventIterator, function() {
315+
if (errorCn > 0) {
316+
console.log("There were errors. Please recheck logs for those.");
317+
}
318+
console.log('finished');
319+
db.close();
320+
db_drill.close();
321+
});
322+
}
236323
});
237324
}
238325
});

‎bin/scripts/expire-data/delete_custom_events.js

+2-6
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@ const common = require('../../../api/utils/common.js');
1212
const drillCommon = require('../../../plugins/drill/api/common.js');
1313

1414
const APP_ID = "";
15-
const EVENTS = []; //leave empty to delete all custom events
15+
const EVENTS = []; //If empty, no events will be deleted
1616

1717
Promise.all([pluginManager.dbConnection("countly"), pluginManager.dbConnection("countly_drill")]).then(async function([countlyDb, drillDb]) {
1818
console.log("Connected to databases...");
@@ -28,11 +28,7 @@ Promise.all([pluginManager.dbConnection("countly"), pluginManager.dbConnection("
2828
//GET EVENTS
2929
var events = EVENTS;
3030
if (!events.length) {
31-
events = await countlyDb.collection("events").findOne({_id: app._id}, {_id: 0, list: 1});
32-
events = (events && events.list) || [];
33-
}
34-
if (!events.length) {
35-
close("No events found");
31+
close("No events to delete");
3632
}
3733
else {
3834
//DELETE EVENTS

0 commit comments

Comments
 (0)