bf: search with aggregate instead of find

The search was using find().sort() and was disrupting user defined
search queries and custom indexes. The sort() is needed to implement
a stateless paging system. The combo of user defined query and sort is
now implemented with a 2 stage aggregate on server side.
We always limit the execution time maxTimeMs to 5mn (tunable by an
environment variable).
The result is staged in a temporary bucket and cached for paging.
We rely on an external job to cleanup the searches (e.g. daily).

Author: vrancurel

lib/storage/metadata/mongoclient/MongoClientInterface.js

@@ -10,6 +10,7 @@
  * We use proper atomic operations when needed.
  */
 const async = require('async');
+const crypto = require('crypto');
 
 const constants = require('../../../constants');
 
@@ -39,6 +40,10 @@ const CONNECT_TIMEOUT_MS = 5000;
 // MongoDB default
 const SOCKET_TIMEOUT_MS = 360000;
 const CONCURRENT_CURSORS = 10;
+// Search
+const MAX_TIME_MS = 300000;
+const TEMP_SEARCH_PREFIX = '__temp_search';
+const SEARCH_PREFIX = '__search';
 
 const initialInstanceID = process.env.INITIAL_INSTANCE_ID;
 
@@ -109,6 +114,11 @@ class MongoClientInterface {
             !Number.isNaN(process.env.CONCURRENT_CURSORS))
             ? Number.parseInt(process.env.CONCURRENT_CURSORS, 10)
             : CONCURRENT_CURSORS;
+
+        this.maxTimeMs = (process.env.MAX_TIME_MS &&
+            !Number.isNaN(process.env.MAX_TIME_MS))
+            ? Number.parseInt(process.env.MAX_TIME_MS, 10)
+            : MAX_TIME_MS;
     }
 
     setup(cb) {
@@ -939,9 +949,10 @@ class MongoClientInterface {
                                       params, log, cb);
     }
 
-    internalListObject(bucketName, params, extension, log, cb) {
-        const c = this.getCollection(bucketName);
-        const stream = new MongoReadStream(c, params, params.mongifiedSearch);
+    internalListObject(c, params, extension, log, cb) {
+        // eslint-disable-next-line
+        params.maxTimeMs = this.maxTimeMs;
+        const stream = new MongoReadStream(c, params, params.searchOptions);
         const skip = new Skip({
             extension,
             gte: params.gte,
@@ -963,7 +974,7 @@ class MongoClientInterface {
             newParams.gte = range;
 
             // then continue listing the next key range
-            this.internalListObject(bucketName, newParams, extension, log, cb);
+            this.internalListObject(c, newParams, extension, log, cb);
         });
 
         stream
@@ -993,22 +1004,125 @@ class MongoClientInterface {
         return undefined;
     }
 
+    /*
+     * Execute the user-defined query in a stage then sort it for
+     * stateless paging. The output is stored in a temporary
+     * collection in a special namespace that will be periodically
+     * erased (e.g. once a day).
+     *
+     * All search queries are bounded by MAX_TIME_MS env (default is
+     * 5mn).
+     */
+    doSearch(c, searchCollection, params, extension, searchOptions, log, cb) {
+        // use temp name to avoid races
+        const tempCollection =
+              TEMP_SEARCH_PREFIX +
+              crypto.randomBytes(16).toString('hex');
+        const _cursor = c.aggregate([
+            { $match: searchOptions }, // user query
+            { $out: tempCollection }, // a job will clean it up
+        ],
+            {
+                maxTimeMs: this.maxTimeMs, // always limit
+                allowDiskUse: true, // stage large queries on disk
+            },
+            null);
+        _cursor.toArray(err => {
+            if (err) {
+                log.error('doSearch: error', {
+                    error: err.message,
+                });
+                return cb(err);
+            }
+            // final rename
+            this.db.renameCollection(
+                tempCollection,
+                searchCollection,
+                {
+                    dropTarget: true,
+                },
+                err => {
+                    if (err) {
+                        log.error('doSearch: renaming', {
+                            error: err.message,
+                            tempCollection,
+                            searchCollection,
+                        });
+                        return cb(err);
+                    }
+                    log.info('doSearch: done', {
+                        searchCollection,
+                    });
+                    return undefined;
+                });
+            // fallthrough
+            // eslint-disable-next-line
+            params.searchOptions = searchOptions;
+            return this.internalListObject(
+                c,
+                params, extension,
+                log, cb);
+        });
+    }
+
+    /*
+     * Check if the used defined query has been cached otherwise
+     * launch the search
+     */
+    prepareSearch(bucketName, params, extension, searchOptions, log, cb) {
+        const c = this.getCollection(bucketName);
+        // generate the search collection name
+        const searchCollection =
+              SEARCH_PREFIX +
+              crypto.createHash('md5').
+              update(JSON.stringify(searchOptions)).
+              digest('hex');
+        this.db.listCollections({
+            name: searchCollection,
+        }).toArray((err, items) => {
+            if (err) {
+                log.error('prepareSearch: listing collection', {
+                    error: err.message,
+                });
+                return cb(err);
+            }
+            if (items.length > 0) {
+                // result is cached
+                return this.internalListObject(
+                    this.db.collection(searchCollection),
+                    params, extension,
+                    log, cb);
+            }
+            return this.doSearch(
+                c, searchCollection,
+                params, extension, searchOptions,
+                log, cb);
+        });
+    }
+
     listObject(bucketName, params, log, cb) {
         const extName = params.listingType;
         const extension = new listAlgos[extName](params, log);
         const internalParams = extension.genMDParams();
-        internalParams.mongifiedSearch = params.mongifiedSearch;
-        return this.internalListObject(bucketName, internalParams, extension,
-                                       log, cb);
+        if (params.mongifiedSearch) {
+            return this.prepareSearch(
+                bucketName, internalParams, extension,
+                params.mongifiedSearch, log, cb);
+        }
+        return this.internalListObject(
+            this.getCollection(bucketName),
+            internalParams, extension,
+            log, cb);
     }
 
     listMultipartUploads(bucketName, params, log, cb) {
         const extName = params.listingType;
         const extension = new listAlgos[extName](params, log);
         const internalParams = extension.genMDParams();
-        internalParams.mongifiedSearch = params.mongifiedSearch;
-        return this.internalListObject(bucketName, internalParams, extension,
-                                       log, cb);
+        return this.internalListObject(
+            this.getCollection(bucketName),
+            internalParams, extension,
+            log, cb);
     }
 
     checkHealth(implName, log, cb) {

lib/storage/metadata/mongoclient/readStream.js

@@ -69,6 +69,9 @@ class MongoReadStream extends Readable {
         if (options.limit && options.limit !== -1) {
             this._cursor = this._cursor.limit(options.limit);
         }
+        if (options.maxTimeMs && options.maxTimeMs !== -1) {
+            this._cursor = this._cursor.maxTimeMS(options.maxTimeMs);
+        }
         this._options = options;
         this._destroyed = false;
         this.on('end', this._cleanup.bind(this));