feat(lifecycle): skip index creation on large collections and mass-expiration rules

extensions/lifecycle/conductor/LifecycleConductor.js

@@ -11,6 +11,7 @@ const BucketClient = require('bucketclient').RESTClient;
 const MongoClient = require('arsenal').storage
     .metadata.mongoclient.MongoClientInterface;
 
+const config = require('../../../lib/Config');
 const BackbeatProducer = require('../../../lib/BackbeatProducer');
 const BackbeatTask = require('../../../lib/tasks/BackbeatTask');
 const ZookeeperManager = require('../../../lib/clients/ZookeeperManager');
@@ -25,6 +26,7 @@ const {
 } = require('../../../lib/constants');
 
 const { LifecycleMetrics } = require('../LifecycleMetrics');
+const { hasMassExpirationRule } = require('../util/mongoRules');
 const { BreakerState, CircuitBreaker } = require('breakbeat').CircuitBreaker;
 const {
     startCircuitBreakerMetricsExport,
@@ -277,6 +279,12 @@ class LifecycleConductor {
                 return cb(null, lifecycleTaskVersions.v1);
             }
 
+            if (task.hasMassExpirationRule) {
+                log.trace('skipping index creation: bucket has a mass-expiration rule');
+                LifecycleMetrics.onLegacyTask(log, 'massExpirationRule');
+                return cb(null, lifecycleTaskVersions.v1);
+            }
+
             if (!this.lcConfig.autoCreateIndexes) {
                 log.trace('skipping index creation: auto creation of indexes disabled');
                 LifecycleMetrics.onLegacyTask(log, 'noIndex');
@@ -331,6 +339,23 @@ class LifecycleConductor {
                     return cb(null, lifecycleTaskVersions.v1);
                 }
 
+                const { maxAutoIndexDocCount, maxAutoIndexStorageBytes } =
+                    config.lifecycleConductorOptions;
+                const docCount = results.collStats.count || 0;
+                const storageSize = results.collStats.storageSize || 0;
+                if (docCount > maxAutoIndexDocCount
+                        || storageSize > maxAutoIndexStorageBytes) {
+                    log.warn('skipping index creation: collection too large', {
+                        bucket: task.bucketName,
+                        docCount,
+                        storageSize,
+                        maxDocCount: maxAutoIndexDocCount,
+                        maxStorageBytes: maxAutoIndexStorageBytes,
+                    });
+                    LifecycleMetrics.onLegacyTask(log, 'collectionTooBig');
+                    return cb(null, lifecycleTaskVersions.v1);
+                }
+
                 this.activeIndexingJobs.push({
                     bucket: task.bucketName,
                     indexes: indexesForFeature.lifecycle.v2,
@@ -893,6 +918,8 @@ class LifecycleConductor {
                                         canonicalId: doc.value.owner,
                                         isLifecycled: !!doc.value.lifecycleConfiguration,
                                         hasSosApi: !!doc.value.capabilities?.VeeamSOSApi,
+                                        hasMassExpirationRule: hasMassExpirationRule(
+                                            doc.value.lifecycleConfiguration?.rules),
                                     });
                                     nEnqueued += 1;
                                     lastSentId = doc._id;

extensions/lifecycle/util/mongoRules.js

@@ -0,0 +1,28 @@
+'use strict';
+
+// Operates on the MongoDB internal lifecycle format (lowercase ruleStatus,
+// actions[], filter.rulePrefix) — NOT the AWS S3 format used by util/rules.js.
+
+// A whole-bucket Expiration with days=0 is the explicit "empty this bucket"
+// signal: treat it as mass-expiration so the conductor keeps lifecycle on v1.
+function isMassExpirationRule(rule) {
+    if (!rule || rule.ruleStatus !== 'Enabled') {
+        return false;
+    }
+    if (rule.filter && (rule.filter.rulePrefix
+            || (rule.filter.tags && rule.filter.tags.length > 0))) {
+        return false;
+    }
+    return (rule.actions || []).some(action => action
+        && action.actionName === 'Expiration'
+        && action.days === 0);
+}
+
+function hasMassExpirationRule(rules) {
+    return (rules || []).some(r => isMassExpirationRule(r));
+}
+
+module.exports = {
+    isMassExpirationRule,
+    hasMassExpirationRule,
+};

lib/Config.js

@@ -180,6 +180,19 @@ class Config extends EventEmitter {
         this.transientLocations = {};
 
         this._setTimeOptions();
+        this._setLifecycleConductorOptions();
+    }
+
+    _setLifecycleConductorOptions() {
+        const maxAutoIndexDocCount =
+            Number.parseInt(process.env.LIFECYCLE_MAX_AUTO_INDEX_DOC_COUNT, 10) || 10000000;
+        const maxAutoIndexStorageBytes =
+            Number.parseInt(process.env.LIFECYCLE_MAX_AUTO_INDEX_STORAGE_BYTES, 10)
+            || 10 * 1024 * 1024 * 1024;
+        this.lifecycleConductorOptions = {
+            maxAutoIndexDocCount,
+            maxAutoIndexStorageBytes,
+        };
     }
 
     _setTimeOptions() {

tests/unit/lifecycle/LifecycleConductor.spec.js

@@ -730,6 +730,116 @@ describe('Lifecycle Conductor', () => {
                 done();
             });
         });
+
+        it('should return v1: bucket has a mass-expiration rule', done => {
+            const client = new BackbeatMetadataProxyMock();
+            conductor.clientManager.getBackbeatMetadataProxy = () => client;
+            conductor.activeIndexingJobsRetrieved = true;
+            conductor.activeIndexingJobs = [];
+            conductor._bucketSource = 'mongodb';
+            client.indexesObj = [];
+            client.error = null;
+
+            const task = { ...getTask(true), hasMassExpirationRule: true };
+            conductor._indexesGetOrCreate(task, log, (err, taskVersion) => {
+                assert.ifError(err);
+                assert.deepStrictEqual(client.receivedIdxObj, null);
+                assert.deepStrictEqual(conductor.activeIndexingJobs, []);
+                assert.deepStrictEqual(taskVersion, lifecycleTaskVersions.v1);
+                done();
+            });
+        });
+
+        it('mass-expiration guard short-circuits before fetching diskUsage/collStats', done => {
+            const client = new BackbeatMetadataProxyMock();
+            conductor.clientManager.getBackbeatMetadataProxy = () => client;
+            conductor.activeIndexingJobsRetrieved = true;
+            conductor.activeIndexingJobs = [];
+            conductor._bucketSource = 'mongodb';
+            // Throw if either MongoDB call is made — the guard must short-circuit before.
+            conductor._mongodbClient = {
+                getDiskUsage: () => { throw new Error('getDiskUsage should not be called'); },
+                getCollectionStats: () => { throw new Error('getCollectionStats should not be called'); },
+            };
+            client.indexesObj = [];
+            client.error = null;
+
+            const task = { ...getTask(true), hasMassExpirationRule: true };
+            conductor._indexesGetOrCreate(task, log, (err, taskVersion) => {
+                assert.ifError(err);
+                assert.deepStrictEqual(taskVersion, lifecycleTaskVersions.v1);
+                done();
+            });
+        });
+
+        it('should still return v2 when mass-expiration bucket already has indexes', done => {
+            const client = new BackbeatMetadataProxyMock();
+            conductor.clientManager.getBackbeatMetadataProxy = () => client;
+            conductor.activeIndexingJobsRetrieved = true;
+            conductor.activeIndexingJobs = [];
+            conductor._bucketSource = 'mongodb';
+            client.indexesObj = indexesForFeature.lifecycle.v2;
+            client.error = null;
+
+            const task = { ...getTask(true), hasMassExpirationRule: true };
+            conductor._indexesGetOrCreate(task, log, (err, taskVersion) => {
+                assert.ifError(err);
+                assert.deepStrictEqual(taskVersion, lifecycleTaskVersions.v2);
+                done();
+            });
+        });
+
+        it('should return v1: collection too big (docCount over threshold)', done => {
+            const client = new BackbeatMetadataProxyMock();
+            conductor.clientManager.getBackbeatMetadataProxy = () => client;
+            conductor.activeIndexingJobsRetrieved = true;
+            conductor.activeIndexingJobs = [];
+            conductor._bucketSource = 'mongodb';
+            conductor._mongodbClient = {
+                getDiskUsage: cb => cb(null, { available: 1e12, free: 1e12, total: 2e12 }),
+                getCollectionStats: (bucketName, _log, cb) => cb(null, {
+                    indexSizes: { _id_: 1000 },
+                    count: 20000000,
+                    storageSize: 100,
+                }),
+            };
+            client.indexesObj = [];
+            client.error = null;
+
+            conductor._indexesGetOrCreate(getTask(true), log, (err, taskVersion) => {
+                assert.ifError(err);
+                assert.deepStrictEqual(client.receivedIdxObj, null);
+                assert.deepStrictEqual(conductor.activeIndexingJobs, []);
+                assert.deepStrictEqual(taskVersion, lifecycleTaskVersions.v1);
+                done();
+            });
+        });
+
+        it('should return v1: collection too big (storageSize over threshold)', done => {
+            const client = new BackbeatMetadataProxyMock();
+            conductor.clientManager.getBackbeatMetadataProxy = () => client;
+            conductor.activeIndexingJobsRetrieved = true;
+            conductor.activeIndexingJobs = [];
+            conductor._bucketSource = 'mongodb';
+            conductor._mongodbClient = {
+                getDiskUsage: cb => cb(null, { available: 1e12, free: 1e12, total: 2e12 }),
+                getCollectionStats: (bucketName, _log, cb) => cb(null, {
+                    indexSizes: { _id_: 1000 },
+                    count: 100,
+                    storageSize: 20 * 1024 * 1024 * 1024,
+                }),
+            };
+            client.indexesObj = [];
+            client.error = null;
+
+            conductor._indexesGetOrCreate(getTask(true), log, (err, taskVersion) => {
+                assert.ifError(err);
+                assert.deepStrictEqual(client.receivedIdxObj, null);
+                assert.deepStrictEqual(conductor.activeIndexingJobs, []);
+                assert.deepStrictEqual(taskVersion, lifecycleTaskVersions.v1);
+                done();
+            });
+        });
     });
 
     describe('listBuckets', () => {
@@ -1101,5 +1211,70 @@ describe('Lifecycle Conductor', () => {
                 done();
             });
         });
+
+        it('should populate hasMassExpirationRule from lifecycle rules when listing from mongodb', done => {
+            const lcConductor = makeLifecycleConductorWithFilters({
+                bucketSource: 'mongodb',
+            }, []);
+
+            const docs = [
+                {
+                    _id: 'mass-exp-bucket',
+                    value: {
+                        owner: 'acc',
+                        lifecycleConfiguration: {
+                            rules: [
+                                { ruleStatus: 'Enabled',
+                                    actions: [{ actionName: 'Expiration', days: 0 }] },
+                            ],
+                        },
+                    },
+                },
+                {
+                    _id: 'selective-bucket',
+                    value: {
+                        owner: 'acc',
+                        lifecycleConfiguration: {
+                            rules: [
+                                { ruleStatus: 'Enabled', actions: [{ actionName: 'Expiration', days: 30 }] },
+                            ],
+                        },
+                    },
+                },
+                { _id: 'no-lifecycle-bucket', value: { owner: 'acc' } },
+            ];
+            let idx = 0;
+            const cursor = {
+                hasNext: () => Promise.resolve(idx < docs.length),
+                next: () => Promise.resolve(docs[idx++]),
+            };
+            const projectStub = sinon.stub().returns(cursor);
+            const findStub = sinon.stub().returns({ project: projectStub });
+
+            lcConductor._mongodbClient = {
+                getIndexingJobs: (_, cb) => cb(null, []),
+                getCollection: name => {
+                    if (name === '__metastore') {return { find: findStub };}
+                    return { collectionName: name };
+                },
+                _isSpecialCollection: () => false,
+            };
+            sinon.stub(lcConductor._zkClient, 'getData').yields(null, null, null);
+
+            lcConductor.listBuckets(queue, fakeLogger, err => {
+                assert.ifError(err);
+                // lifecycleConfiguration must be in the projection so we can read its rules
+                assert.strictEqual(projectStub.getCall(0).args[0]['value.lifecycleConfiguration'], 1);
+                const tasks = queue.list();
+                assert.strictEqual(tasks.length, 3);
+                const massExp = tasks.find(t => t.bucketName === 'mass-exp-bucket');
+                const selective = tasks.find(t => t.bucketName === 'selective-bucket');
+                const noLifecycle = tasks.find(t => t.bucketName === 'no-lifecycle-bucket');
+                assert.strictEqual(massExp.hasMassExpirationRule, true);
+                assert.strictEqual(selective.hasMassExpirationRule, false);
+                assert.strictEqual(noLifecycle.hasMassExpirationRule, false);
+                done();
+            });
+        });
     });
 });