replication rules - actual delete by db cleaner dfs 2353 (#9024)

Signed-off-by: Amit Prinz Setter <alphaprinz@gmail.com>

Author: alphaprinz

src/server/bg_services/db_cleaner.js

@@ -8,6 +8,7 @@ const config = require('../../../config');
 const MDStore = require('../object_services/md_store').MDStore;
 const system_store = require('../system_services/system_store').get_instance();
 const nodes_store = require('../node_services/nodes_store').NodesStore.instance();
+const replication_store = require('../system_services/replication_store').instance();
 const system_utils = require('../utils/system_utils');
 const db_client = require('../../util/db_client');
 const md_aggregator = require('./md_aggregator');
@@ -47,6 +48,7 @@ async function background_worker() {
     await clean_md_store(last_date_to_remove);
     await clean_nodes_store(last_date_to_remove);
     await clean_system_store(last_date_to_remove);
+    await clean_replication_store(last_date_to_remove);
     dbg.log0('DB_CLEANER:', 'END');
     return config.DB_CLEANER_CYCLE;
 }
@@ -107,6 +109,25 @@ async function clean_nodes_store(last_date_to_remove) {
     dbg.log0(`DB_CLEANER: removed ${filtered_nodes.length} documents from nodes-store`);
 }
 
+async function clean_replication_store(last_date_to_remove) {
+    const total_replication_rules_count = await replication_store.count_total_replication_rules();
+    if (total_replication_rules_count < config.DB_CLEANER_MAX_TOTAL_DOCS) {
+        dbg.log0(`DB_CLEANER: found less than ${config.DB_CLEANER_MAX_TOTAL_DOCS} replication rules in replication-store
+        ${total_replication_rules_count} replication rules - Skipping...`);
+        return;
+    }
+    dbg.log0('DB_CLEANER: checking replication-store for replication rules deleted before', new Date(last_date_to_remove));
+    const replication_rules = await replication_store.find_deleted_rules(last_date_to_remove, config.DB_CLEANER_DOCS_LIMIT);
+    if (replication_rules.length === 0) {
+        dbg.log0("No replication rules to delete.");
+        return;
+    }
+    const rr_ids = db_client.instance().uniq_ids(replication_rules, '_id');
+    dbg.log0('DB_CLEANER: found deleted replication rules:', rr_ids);
+    await replication_store.actual_delete_replication_by_id(rr_ids);
+    dbg.log0(`DB_CLEANER: removed ${rr_ids.length} replication rules from replication-store`);
+}
+
 async function clean_system_store(last_date_to_remove) {
     const total_accounts_count = await system_store.count_total_docs('accounts');
     const total_buckets_count = await system_store.count_total_docs('buckets');

src/server/system_services/bucket_server.js

@@ -963,7 +963,7 @@ async function delete_bucket_and_objects(req) {
 
     if (bucket.replication_policy_id) {
         // delete replication from replication collection
-        await replication_store.instance().delete_replication_by_id(bucket.replication_policy_id);
+        await replication_store.instance().mark_deleted_replication_by_id(bucket.replication_policy_id);
     }
 
     const req_account = req.account &&
@@ -1027,7 +1027,7 @@ async function delete_bucket(req) {
         });
         if (bucket.replication_policy_id) {
             // delete replication from replication collection
-            await replication_store.instance().delete_replication_by_id(bucket.replication_policy_id);
+            await replication_store.instance().mark_deleted_replication_by_id(bucket.replication_policy_id);
         }
 
         await BucketStatsStore.instance().delete_stats({
@@ -2079,7 +2079,7 @@ async function delete_bucket_replication(req) {
     });
 
     // delete replication from replication collection
-    await replication_store.instance().delete_replication_by_id(replication_id);
+    await replication_store.instance().mark_deleted_replication_by_id(replication_id);
 }
 
 async function validate_replication(req) {

src/server/system_services/replication_store.js

@@ -59,14 +59,23 @@ class ReplicationStore {
         return repl;
     }
 
+    async find_deleted_rules(last_date_to_remove, limit) {
+        const repl = this._replicationconfigs.find({
+            deleted: {
+                $lt: last_date_to_remove
+            },
+        }, { limit, projection: { _id: 1, deleted: 1 } });
+        return repl;
+    }
+
     async get_replication_by_id(replication_id) {
         dbg.log1('get_replication_by_id: ', replication_id);
         const repl = await this._replicationconfigs.findOne({ _id: db_client.instance().parse_object_id(replication_id), deleted: null });
         return repl && repl.rules;
     }
 
-    async delete_replication_by_id(_id) {
-        dbg.log1('delete_replication_by_id: ', _id);
+    async mark_deleted_replication_by_id(_id) {
+        dbg.log0('mark_deleted_replication_by_id: ', _id);
         const ans = await this._replicationconfigs.updateOne({
             _id: db_client.instance().parse_object_id(_id),
             deleted: null
@@ -78,6 +87,17 @@ class ReplicationStore {
         return ans;
     }
 
+    async actual_delete_replication_by_id(ids) {
+        dbg.log0('actual_delete_replication_by_ids: ', ids);
+        if (!ids || !ids.length) return;
+        return this._replicationconfigs.deleteMany({
+            _id: {
+                $in: ids
+            },
+            deleted: { $exists: true }
+        });
+    }
+
     async update_replication_status_by_id(_id, rule_id, status) {
         dbg.log1('update_replication_status_by_id: ', _id, rule_id, status);
         const parsed_id = db_client.instance().parse_object_id(_id);
@@ -153,6 +173,10 @@ class ReplicationStore {
         return reduced_replications;
     }
 
+    async count_total_replication_rules() {
+        return this._replicationconfigs.countDocuments({});
+    }
+
 }
 
 // EXPORTS