Skip to content

Commit 9ab68b6

Browse files
pnrgencpnrgenc
authored
Merge pull request #6993 from Countly/adding-report-to-mutation-manager-job
feat: MutationManagerJob improvements
2 parents fc3d5a8 + 7e61ea3 commit 9ab68b6

2 files changed

Lines changed: 161 additions & 19 deletions

File tree

‎api/api.js‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -161,6 +161,26 @@ plugins.connectToAllDatabases().then(function() {
161161
}
162162
);
163163

164+
// mutation manager default settings
165+
[
166+
{ key: 'max_retries', value: 3 },
167+
{ key: 'retry_delay_ms', value: 30 * 60 * 1000 },
168+
{ key: 'validation_interval_ms', value: 3 * 60 * 1000 },
169+
{ key: 'stale_ms', value: 24 * 60 * 60 * 1000 },
170+
{ key: 'batch_limit', value: 10 }
171+
].forEach(item => {
172+
const path = `mutation_manager.${item.key}`;
173+
const update = {};
174+
update[path] = item.value;
175+
common.db.collection('plugins').updateOne(
176+
{ _id: 'plugins', [path]: { $exists: false } },
177+
{ $set: update }
178+
).catch(e => {
179+
const message = `Failed to add mutation_manager default for ${item.key}: ${e}`;
180+
log && log.e ? log.e(message) : console.log(message);
181+
});
182+
});
183+
164184
if (common.queryRunner && typeof common.queryRunner.isAdapterAvailable === 'function' && common.queryRunner.isAdapterAvailable('clickhouse')) {
165185
common.db.collection('plugins').updateOne(
166186
{

‎api/jobs/mutationManagerJob.js‎

Lines changed: 141 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,17 @@ const common = require('../utils/common.js');
22
const log = require("../utils/log.js")("job:mutationManager");
33
const Job = require("../../jobServer/Job.js");
44
const mutationManager = require('../utils/mutationManager.js');
5+
const tracker = require("../parts/mgmt/tracker.js");
6+
const plugins = require('../../plugins/pluginManager.js');
7+
8+
const DEFAULT_JOB_CONFIG = {
9+
STALE_MS: 24 * 60 * 60 * 1000, // 24h - consider tasks running longer than this as stale
10+
RETRY_DELAY_MS: 30 * 60 * 1000, // 30m - delay before retrying a failed task
11+
MAX_RETRIES: 3, // Max number of retries before marking a task as failed
12+
VALIDATION_INTERVAL_MS: 3 * 60 * 1000, // 3m - interval between mutation status checks
13+
BATCH_LIMIT: 10 // Number of tasks to process in one run
14+
};
15+
let jobConfigState = { ...DEFAULT_JOB_CONFIG };
516

617
let clickHouseRunner;
718
try {
@@ -19,14 +30,43 @@ catch {
1930
//
2031
}
2132

33+
/**
34+
* Normalize mutation manager job configuration values
35+
* @param {Object} cfg - Raw configuration object from the DB
36+
* @returns {Object} Normalized configuration
37+
*/
38+
function buildJobConfig(cfg = {}) {
39+
return {
40+
STALE_MS: common.isNumber(cfg.stale_ms) ? Number(cfg.stale_ms) : DEFAULT_JOB_CONFIG.STALE_MS,
41+
RETRY_DELAY_MS: common.isNumber(cfg.retry_delay_ms) ? Number(cfg.retry_delay_ms) : DEFAULT_JOB_CONFIG.RETRY_DELAY_MS,
42+
MAX_RETRIES: common.isNumber(cfg.max_retries) ? Number(cfg.max_retries) : DEFAULT_JOB_CONFIG.MAX_RETRIES,
43+
VALIDATION_INTERVAL_MS: common.isNumber(cfg.validation_interval_ms) ? Number(cfg.validation_interval_ms) : DEFAULT_JOB_CONFIG.VALIDATION_INTERVAL_MS,
44+
BATCH_LIMIT: common.isNumber(cfg.batch_limit) ? Number(cfg.batch_limit) : DEFAULT_JOB_CONFIG.BATCH_LIMIT
45+
};
46+
}
2247

23-
// Constants for managing stale tasks and retries
24-
const STALE_MS = 24 * 60 * 60 * 1000; // 24h - consider tasks running longer than this as stale
25-
const RETRY_DELAY_MS = 30 * 60 * 1000; // 30m - delay before retrying a failed task
26-
const MAX_RETRIES = 3; // Max number of retries before marking a task as failed
27-
const VALIDATION_INTERVAL_MS = 3 * 60 * 1000; // 3m - interval between mutation status checks
48+
/**
49+
* Fetch mutation manager job configuration from the plugins collection
50+
* @returns {Promise<Object>} Job configuration
51+
*/
52+
async function loadMutationManagerJobConfig() {
53+
if (!common.db) {
54+
return { ...DEFAULT_JOB_CONFIG };
55+
}
2856

29-
const BATCH_LIMIT = 10; // Number of tasks to process in one run
57+
try {
58+
const doc = await common.db.collection('plugins').findOne(
59+
{ _id: 'plugins' },
60+
{ projection: { 'mutation_manager.max_retries': 1, 'mutation_manager.retry_delay_ms': 1, 'mutation_manager.validation_interval_ms': 1, 'mutation_manager.stale_ms': 1, 'mutation_manager.batch_limit': 1 } }
61+
);
62+
const cfg = (doc && doc.mutation_manager) || {};
63+
return buildJobConfig(cfg);
64+
}
65+
catch (e) {
66+
log.e("Failed to load mutation manager job config; using defaults", e?.message || e);
67+
return { ...DEFAULT_JOB_CONFIG };
68+
}
69+
}
3070

3171
/** Class for the mutation manager job **/
3272
class MutationManagerJob extends Job {
@@ -54,6 +94,8 @@ class MutationManagerJob extends Job {
5494
* @param {done} done callback
5595
*/
5696
async run() {
97+
jobConfigState = await loadMutationManagerJobConfig();
98+
const jobConfig = jobConfigState || DEFAULT_JOB_CONFIG;
5799
const now = Date.now();
58100
const batchId = common.db.ObjectID() + '';
59101
const summary = [];
@@ -71,7 +113,7 @@ class MutationManagerJob extends Job {
71113
}
72114
},
73115
{ $sort: { ts: 1 } },
74-
{ $limit: BATCH_LIMIT },
116+
{ $limit: jobConfig.BATCH_LIMIT },
75117
{ $set: { running: true, status: mutationManager.MUTATION_STATUS.RUNNING, hb: now, error: null, batch_id: batchId } },
76118
{
77119
$merge: {
@@ -111,8 +153,9 @@ class MutationManagerJob extends Job {
111153
* Process a single mutation task (delete/update)
112154
* @param {Object} task - Task document
113155
* @param {Array} summary - Summary array to push statuses
156+
* @param {Object} [jobConfig] - Job configuration
114157
*/
115-
async processTask(task, summary) {
158+
async processTask(task, summary, jobConfig = jobConfigState || DEFAULT_JOB_CONFIG) {
116159
const type = task.type;
117160
if (type !== 'delete' && type !== 'update') {
118161
await common.db.collection("mutation_manager").updateOne(
@@ -138,7 +181,7 @@ class MutationManagerJob extends Job {
138181
const now = Date.now();
139182
await common.db.collection('mutation_manager').updateOne(
140183
{ _id: task._id },
141-
{ $set: { running: false, status: mutationManager.MUTATION_STATUS.QUEUED, hb: now, error: pre.reason || 'deferred_due_to_ch_pressure', retry_at: now + RETRY_DELAY_MS }, $unset: { batch_id: '' } }
184+
{ $set: { running: false, status: mutationManager.MUTATION_STATUS.QUEUED, hb: now, error: pre.reason || 'deferred_due_to_ch_pressure', retry_at: now + jobConfig.RETRY_DELAY_MS }, $unset: { batch_id: '' } }
142185
);
143186
summary.push({ query: task.query, status: 'deferred_due_to_ch_pressure', reason: pre.reason });
144187
return;
@@ -173,7 +216,7 @@ class MutationManagerJob extends Job {
173216
running: false,
174217
status: mutationManager.MUTATION_STATUS.AWAITING_CH_MUTATION_VALIDATION,
175218
hb: Date.now(),
176-
retry_at: Date.now() + VALIDATION_INTERVAL_MS
219+
retry_at: Date.now() + jobConfig.VALIDATION_INTERVAL_MS
177220
},
178221
$unset: { batch_id: "" }
179222
}
@@ -204,9 +247,10 @@ class MutationManagerJob extends Job {
204247
* Processes a batch of tasks awaiting ClickHouse mutation validation.
205248
* Reserves a batch, checks mutation status via system.mutations and updates/deletes tasks accordingly.
206249
* @param {Array} summary - Summary array to push status logs
250+
* @param {Object} [jobConfig] - Job configuration
207251
* @returns {Promise<void>} Resolves when awaiting validation batch is processed
208252
*/
209-
async processAwaitingValidation(summary) {
253+
async processAwaitingValidation(summary, jobConfig = jobConfigState || DEFAULT_JOB_CONFIG) {
210254
const validationBatchId = common.db.ObjectID() + '';
211255
const nowTs = Date.now();
212256
await common.db.collection("mutation_manager").aggregate([
@@ -218,7 +262,7 @@ class MutationManagerJob extends Job {
218262
}
219263
},
220264
{ $sort: { ts: 1 } },
221-
{ $limit: BATCH_LIMIT },
265+
{ $limit: jobConfig.BATCH_LIMIT },
222266
{ $set: { running: true, hb: nowTs, batch_id: validationBatchId } },
223267
{
224268
$merge: {
@@ -268,7 +312,7 @@ class MutationManagerJob extends Job {
268312
await common.db.collection("mutation_manager").updateOne(
269313
{ _id: task._id },
270314
{
271-
$set: { running: false, status: mutationManager.MUTATION_STATUS.AWAITING_CH_MUTATION_VALIDATION, hb: Date.now(), retry_at: Date.now() + VALIDATION_INTERVAL_MS },
315+
$set: { running: false, status: mutationManager.MUTATION_STATUS.AWAITING_CH_MUTATION_VALIDATION, hb: Date.now(), retry_at: Date.now() + jobConfig.VALIDATION_INTERVAL_MS },
272316
$unset: { batch_id: "" }
273317
}
274318
);
@@ -285,14 +329,15 @@ class MutationManagerJob extends Job {
285329

286330
/**
287331
* Reset stale mutation tasks that have been running longer than STALE_MS.
332+
* @param {Object} [jobConfig] - Job configuration
288333
*/
289-
async resetStaleTasks() {
334+
async resetStaleTasks(jobConfig = jobConfigState || DEFAULT_JOB_CONFIG) {
290335
try {
291336
const now = Date.now();
292337
await common.db.collection("mutation_manager").updateMany(
293338
{
294339
running: true,
295-
$expr: { $lt: [{ $ifNull: ["$hb", "$ts"] }, now - STALE_MS] }
340+
$expr: { $lt: [{ $ifNull: ["$hb", "$ts"] }, now - jobConfig.STALE_MS] }
296341
},
297342
{
298343
$set: { running: false, status: mutationManager.MUTATION_STATUS.QUEUED, hb: now, error: "stale_reset" },
@@ -475,25 +520,34 @@ class MutationManagerJob extends Job {
475520
* Marks a task as failed or schedules it for a retry based on the number of previous failures.
476521
* @param {Object} task - The task object to update.
477522
* @param {string} message - The error message to log
523+
* @param {Object} [jobConfig] - Job configuration
478524
*/
479-
async markFailedOrRetry(task, message) {
525+
async markFailedOrRetry(task, message, jobConfig = jobConfigState || DEFAULT_JOB_CONFIG) {
480526
try {
481527
const now = Date.now();
482528
const failCount = (task.fail_count || 0) + 1;
483-
if (failCount >= MAX_RETRIES) {
529+
if (failCount >= jobConfig.MAX_RETRIES) {
484530
await common.db.collection("mutation_manager").updateOne(
485531
{ _id: task._id },
486532
{
487533
$set: { running: false, status: mutationManager.MUTATION_STATUS.FAILED, fail_count: failCount, error: message, hb: now },
488534
$unset: { batch_id: "" }
489535
}
490536
);
537+
await this.reportFailureToStats(
538+
task,
539+
'Some records could not be deleted and the maximum retry limit has been reached. Please review these records, otherwise they will remain on the server. ' +
540+
message +
541+
(
542+
task.error ? ` | Task error: ${String(task.error).slice(0, 900)}` : ''
543+
)
544+
);
491545
}
492546
else {
493547
await common.db.collection("mutation_manager").updateOne(
494548
{ _id: task._id },
495549
{
496-
$set: { running: false, status: mutationManager.MUTATION_STATUS.QUEUED, fail_count: failCount, error: message, retry_at: now + RETRY_DELAY_MS, hb: now },
550+
$set: { running: false, status: mutationManager.MUTATION_STATUS.QUEUED, fail_count: failCount, error: message, retry_at: now + jobConfig.RETRY_DELAY_MS, hb: now },
497551
$unset: { batch_id: "" }
498552
}
499553
);
@@ -564,6 +618,74 @@ class MutationManagerJob extends Job {
564618
}
565619
}
566620
}
567-
}
568621

622+
/**
623+
* Ensure tracker is enabled
624+
* @returns {boolean} if enabled
625+
*/
626+
ensureTracker() {
627+
try {
628+
if (!tracker || typeof tracker.isEnabled === 'undefined') {
629+
return false;
630+
}
631+
632+
const trackingCfg = plugins.getConfig && plugins.getConfig('tracking');
633+
if (trackingCfg && trackingCfg.server_crashes === false) {
634+
return false;
635+
}
636+
637+
if (tracker.isEnabled()) {
638+
return true;
639+
}
640+
641+
if (typeof tracker.enable !== 'undefined') {
642+
tracker.enable();
643+
}
644+
return tracker.isEnabled();
645+
}
646+
catch (e) {
647+
log.e('ensureTracker failed', e?.message || e + "");
648+
return false;
649+
}
650+
}
651+
652+
/**
653+
* Report failed mutations to stats (crashes/new hook)
654+
* @param {Object} task - The mutation task
655+
* @param {string} message - Failure reason
656+
*/
657+
async reportFailureToStats(task, message) {
658+
try {
659+
if (!this.ensureTracker()) {
660+
return;
661+
}
662+
663+
const Countly = tracker.getSDK && tracker.getSDK();
664+
const payload = {
665+
message: (message || '').toString().slice(0, 900),
666+
db: task && task.db || '',
667+
collection: task && task.collection || '',
668+
type: task && task.type || '',
669+
task_id: task && task._id ? String(task._id) : '',
670+
error: task && task.error ? String(task.error).slice(0, 900) : '',
671+
fail_count: String(task && task.fail_count || 0)
672+
};
673+
674+
if (task && task.query) {
675+
try {
676+
payload.query = JSON.stringify(task.query).slice(0, 900);
677+
}
678+
catch (e) {
679+
payload.query = '[unserializable_query]';
680+
}
681+
}
682+
if (Countly && typeof Countly.log_error !== 'undefined') {
683+
Countly.log_error(new Error(`mutation_manager_job_failed: ${payload.message || 'unknown'}`), payload);
684+
}
685+
}
686+
catch (e) {
687+
log.e("reportFailureToStats failed", e?.message || e + "");
688+
}
689+
}
690+
}
569691
module.exports = MutationManagerJob;

0 commit comments

Comments
 (0)