Merge pull request #6838 from Countly/deletion-manager-observability

fix: restructuring

Author: pnrgenc

api/jobs/deletionManagerJob.js

@@ -49,16 +49,6 @@ class DeletionManagerJob extends Job {
         return 1;
     }
 
-    /**
-     * Check if ClickHouse is enabled
-     * @returns {boolean} true if ClickHouse is enabled, false otherwise
-     */
-    isClickhouseEnabled() {
-        return !!(common.queryRunner
-        && typeof common.queryRunner.isAdapterAvailable === 'function'
-        && common.queryRunner.isAdapterAvailable('clickhouse'));
-    }
-
     /**
      * Run the job
      * @param {done} done callback
@@ -69,7 +59,7 @@ class DeletionManagerJob extends Job {
         const summary = [];
 
         try {
-            if (this.isClickhouseEnabled()) {
+            if (deletionManager.isClickhouseEnabled()) {
                 const pre = await this.shouldDeferDueToClickhousePressure();
                 if (pre && pre.defer) {
                     log.d("Run deferred due to ClickHouse pressure", pre || {});
@@ -120,7 +110,7 @@ class DeletionManagerJob extends Job {
                 if (task.db === "countly_drill" && task.collection === "drill_events") {
                     const mongoOk = await this.deleteMongo(task);
                     let chScheduledOk = true;
-                    const clickhouseEnabled = this.isClickhouseEnabled();
+                    const clickhouseEnabled = deletionManager.isClickhouseEnabled();
                     const hasClickhouse = clickhouseEnabled && !!(clickHouseRunner && clickHouseRunner.deleteGranularDataByQuery);
                     if (hasClickhouse) {
                         chScheduledOk = await this.deleteClickhouse(task);
@@ -167,7 +157,7 @@ class DeletionManagerJob extends Job {
             }
         }
 
-        if (this.isClickhouseEnabled()) {
+        if (deletionManager.isClickhouseEnabled()) {
             await this.processAwaitingValidation(summary);
         }
 
@@ -211,8 +201,8 @@ class DeletionManagerJob extends Job {
 
         for (const task of awaiting) {
             try {
-                if (task.db === "countly_drill" && task.collection === "drill_events" && clickHouseRunner && clickHouseRunner.getMutationStatus) {
-                    const status = await this.getClickhouseMutationStatus(task);
+                if (task.db === "countly_drill" && task.collection === "drill_events" && chHealth && typeof chHealth.getMutationStatus === 'function') {
+                    const status = await chHealth.getMutationStatus({ validation_command_id: task.validation_command_id, table: task.collection, database: task.db });
                     if (status && status.is_done) {
                         await common.db.collection("deletion_manager").updateOne(
                             { _id: task._id },
@@ -355,34 +345,6 @@ class DeletionManagerJob extends Job {
         }
     }
 
-    /**
-     * Retrieves ClickHouse mutation status for a task via system.mutations.
-     * @param {object} task - Deletion task
-     * @returns {Promise<object|null>} Object with keys: is_done:boolean, is_killed:boolean, latest_fail_reason:string|null
-     */
-    async getClickhouseMutationStatus(task) {
-        if (!common.queryRunner) {
-            return null;
-        }
-        const commandId = task.validation_command_id;
-        const queryDef = {
-            name: 'GET_MUTATION_STATUS',
-            adapters: {
-                clickhouse: {
-                    handler: clickHouseRunner.getMutationStatus
-                }
-            }
-        };
-        try {
-            const res = await common.queryRunner.executeQuery(queryDef, { validation_command_id: commandId, table: task.collection, database: task.db }, {});
-            return res ? { is_done: !!res.is_done, is_killed: !!res.is_killed, latest_fail_reason: res.latest_fail_reason || null } : null;
-        }
-        catch (err) {
-            log.e("getClickhouseMutationStatus failed", { taskId: task._id, error: err?.message || err + "" });
-            return null;
-        }
-    }
-
     /**
      * Marks a task as failed or schedules it for a retry based on the number of previous failures.
      * @param {Object} task - The task object to update.
@@ -428,20 +390,15 @@ class DeletionManagerJob extends Job {
      * @returns {Promise<Object|null>} Object with metrics or null if fetch failed
      */
     async getClickhouseLoadMetrics() {
-        if (!common.queryRunner || !clickHouseRunner || !clickHouseRunner.getClickhouseHealthMetrics) {
+        if (!chHealth || typeof chHealth.getOperationalSnapshot !== 'function') {
             return null;
         }
         try {
-            const queryDef = {
-                name: 'GET_CH_HEALTH_METRICS',
-                adapters: {
-                    clickhouse: {
-                        handler: clickHouseRunner.getClickhouseHealthMetrics
-                    }
-                }
-            };
-            const res = await common.queryRunner.executeQuery(queryDef, {}, {});
-            return res || null;
+            const snapshot = await chHealth.getOperationalSnapshot({ database: 'countly_drill', table: 'drill_events' });
+            return snapshot ? {
+                max_parts_per_partition: snapshot.max_parts_per_partition,
+                total_merge_tree_parts: snapshot.total_merge_tree_parts
+            } : null;
         }
         catch (e) {
             log.e('CH health metrics fetch failed', e?.message || e + "");

api/utils/deletionManager.js

@@ -3,12 +3,12 @@ const common = require('./common.js'),
     plugins = require('../../plugins/pluginManager.js'),
     manager = {};
 
-let clickHouseRunner;
+let chHealth = null;
 try {
-    clickHouseRunner = require('../../plugins/clickhouse/api/queries/clickhouseCoreQueries.js');
+    chHealth = require('../../plugins/clickhouse/api/health.js');
 }
-catch (e) {
-    clickHouseRunner = null;
+catch {
+    //
 }
 
 (function() {
@@ -20,6 +20,16 @@ catch (e) {
         DELETED: "deleted"
     };
 
+    /**
+     * Check if ClickHouse is enabled
+     * @returns {boolean} true if ClickHouse is enabled, false otherwise
+     */
+    manager.isClickhouseEnabled = function() {
+        return !!(common.queryRunner
+            && typeof common.queryRunner.isAdapterAvailable === 'function'
+            && common.queryRunner.isAdapterAvailable('clickhouse'));
+    };
+
     plugins.register("/master", function() {
         common.db.collection('deletion_manager').ensureIndex({"deletion_completion_ts": 1}, {expireAfterSeconds: 3 * 24 * 60 * 60}, function() {});
     });
@@ -50,23 +60,27 @@ catch (e) {
     plugins.register('/system/observability/collect', async function() {
         try {
             const summary = await getQueueSummary();
-            const mutations = await getPendingMutationsFromCH();
+
+            const metrics = {
+                summary: {
+                    queued: summary.queued,
+                    running: summary.running,
+                    awaiting_validation: summary.awaiting_validation,
+                    failed: summary.failed,
+                    deleted: summary.deleted,
+                    oldest_wait_sec: summary.oldest_wait_sec
+                }
+            };
+
+            if (manager.isClickhouseEnabled()) {
+                metrics.mutations = await getPendingMutationsFromCH();
+            }
 
             return {
                 provider: 'deletion',
                 healthy: summary.failed === 0,
                 issues: [],
-                metrics: {
-                    summary: {
-                        queued: summary.queued,
-                        running: summary.running,
-                        awaiting_validation: summary.awaiting_validation,
-                        failed: summary.failed,
-                        deleted: summary.deleted,
-                        oldest_wait_sec: summary.oldest_wait_sec
-                    },
-                    mutations
-                },
+                metrics,
                 date: new Date().toISOString()
             };
         }
@@ -134,41 +148,22 @@ catch (e) {
      * @returns {Promise<Object>} - Operational snapshot
      */
     async function getPendingMutationsFromCH() {
-        if (!clickHouseRunner) {
-            return { pending: 0, details: [], error: 'clickHouseRunner_not_available'};
-        }
-        const res = await clickHouseRunner.listPendingMutations({ database: 'countly_drill', table: 'drill_events' });
-        const rows = res && res.data ? res.data : [];
-        return {
-            pending: rows.length,
-            details: rows.map(r => ({
-                command: r.command + '',
-                is_killed: r.is_killed === 1 || r.is_killed === '1',
-                latest_fail_reason: r.latest_fail_reason || null
-            }))
-        };
-    }
-
-    manager.getDeletionClickhousePressureLimits = async function() {
         try {
-            const doc = await common.db.collection('plugins').findOne(
-                { _id: 'plugins' },
-                { projection: { 'deletion_manager.ch_max_parts_per_partition': 1, 'deletion_manager.ch_max_total_mergetree_parts': 1 } }
-            );
-            const dmCfg = (doc && doc.deletion_manager) ? doc.deletion_manager : {};
+            const rows = await chHealth.listPendingMutations({ database: 'countly_drill', table: 'drill_events' });
+            const list = Array.isArray(rows) ? rows : [];
             return {
-                CH_MAX_PARTS_PER_PARTITION: Number(dmCfg.ch_max_parts_per_partition) || 1000,
-                CH_MAX_TOTAL_MERGETREE_PARTS: Number(dmCfg.ch_max_total_mergetree_parts) || 100000
+                pending: list.length,
+                details: list.map(r => ({
+                    command: r.command + '',
+                    is_killed: r.is_killed === 1 || r.is_killed === '1',
+                    latest_fail_reason: r.latest_fail_reason || null
+                }))
             };
         }
         catch (e) {
-            log.e('Failed to load deletion manager thresholds from DB, using defaults', e && e.message ? e.message : e);
-            return {
-                CH_MAX_PARTS_PER_PARTITION: 1000,
-                CH_MAX_TOTAL_MERGETREE_PARTS: 100000
-            };
+            return { pending: 0, details: [], error: 'clickhouse_health_unavailable' };
         }
-    };
+    }
 })(manager);
 
-module.exports = manager;
+module.exports = manager;