Source: libs/change_streams.js

  1. /**
  2.  * Change Streams library.
  3.  */
  5. /*global
  6.  assert CountDownLatch Date db Error Mongo print ScopedThread sharded_cluster sleep tojson
  7. */
  10. /**
  11.  * Creates a ChangeStreamListenerThread.
  12.  *
  13.  * @constructor
  14.  * @param {String} dbName - The database name.
  15.  * @param {String} collName - The collection name.
  16.  * @param {Array} filterPipeline - The aggregation pipeline that should follow the '$changeStream'
  17.  *        aggregation stage.
  18.  * @param {Object} changeStreamOptions - The options that should be passed to the '$changeStream'
  19.  *        aggregation stage. Defaults to no options.
  20.  * @param {Function} changeHandler - A callback function that takes a change document, a state
  21.  *        object and an options object as parameters. The function will be called for each change
  22.  *        document read. Defaults to a noop function.
  23.  * @param {Object} changeHandlerOptions - An object that will be passed as argument to the
  24.  *        changeHandler callback function. Defaults to an empty object.
  25.  * @param {String} targetHost - The host to connect to in order to listen for the change stream.
  26.  *        Defaults to the the default server (using the global db).
  27.  * @param {String} readPreferenceMode - The read preference of the change stream aggregation pipeline.
  28.  */
  29. var ChangeStreamListenerThread = function(dbName, collName, filterPipeline, changeStreamOptions,
  30.                                           changeHandler, changeHandlerOptions, targetHost, readPreferenceMode) {
  31.     var _continueCounter = null;
  32.     var _notReadyCounter = null;
  33.     var _thread = null;
  34.     var _filterPipeline = filterPipeline || [];
  35.     var _changeStreamOptions = changeStreamOptions || {};
  36.     var _changeHandler = changeHandler || function (doc) {};
  37.     var _changeHandlerOptions = changeHandlerOptions || {};
  39.     /**
  40.      * Create a change stream cursor and continually read from it.
  41.      * This function is executed in a ScopedThread.
  42.      *
  43.      * The 'continueCounter' parameter is a CountDownLatch used by the main thread to signal this
  44.      * function, which is executed in a ScopedThread, if it can continue or should stop. Its count
  45.      * must be set to 1 to indicate that the function can continue, and to 0 to indicate that it
  46.      * should stop.
  47.      * The 'notReadyCounter' parameter is a CountDownLatch used by this function to signal the main
  48.      * thread that it is ready and reading change streams. It must be passed to this function with
  49.      * a count of 1, and the function will set it to 0 when ready.
  50.      */
  51.     function _readChangeStream(continueCounter, notReadyCounter, dbName, collName, filterPipeline,
  52.                                changeStreamOptions, changeHandler, changeHandlerOptions,
  53.                                targetHost, username, password, authEnabled, readPreferenceMode) {
  54.         try {
  55.             assert.eq(continueCounter.getCount(), 1,
  56.                       "The continueCounter must have an initial value of 1.");
  57.             assert.eq(notReadyCounter.getCount(), 1,
  58.                       "The notReadyCounter must have an initial value of 1.");
  59.             var state = {
  60.                 "value": 0,
  61.                 "clusterTime": null
  62.             };
  63.             var pipeline = [{"$changeStream": changeStreamOptions}].concat(filterPipeline);
  65.             // Initialize the db object.
  66.             var _db;
  67.             if (targetHost) {
  68.                 // Using a connection to the provided 'targetHost'.
  69.                 // This is used when we need to connect to a specific mongos.
  70.                 if (authEnabled){
  71.                     var targetHostAuth = 'mongodb://'.concat(username, ':', password, '@', targetHost);
  72.                 } else {
  73.                     var targetHostAuth = targetHost;
  74.                 }
  75.                 var conn = new Mongo(targetHostAuth);
  76.                 _db = conn.getDB(dbName);
  77.             } else {
  78.                 // Using the default connection configured by run_workloads.py.
  79.                 _db = db.getSiblingDB(dbName);
  80.             }
  82.             // Function to execute a command and update the state.clusterTime variable with
  83.             // the cluster time found in the response.
  84.             var runCommand = function(cmdObj) {
  85.                 var res = assert.commandWorked(_db.runCommand(cmdObj));
  86.                 state.clusterTime = res.$clusterTime.clusterTime.getTime();
  87.                 return res;
  88.             };
  90.             // Run the aggregate command.
  91.             var res = runCommand({aggregate: collName,
  92.                                   pipeline: pipeline,
  93.                                   cursor: {},
  94.                                   $readPreference: {mode: readPreferenceMode}});
  95.             var cursorId = res.cursor.id;
  96.             var batch = res.cursor.firstBatch;
  98.             // Set up the function that will be called for each document.
  99.             var callback = function(doc) {
  100.                 changeHandler(doc, state, changeHandlerOptions);
  101.             };
  103.             // The aggregation command returned sucessfully. This thread is ready.
  104.             notReadyCounter.countDown();
  105.             // While the thread is not stopped, we go through each batch of result, call the
  106.             // callback function for each document, and fetch the next batch with a 'getMore'
  107.             // command.
  108.             // Note that we don't wait for all the changes to be read as for some configurations
  109.             // it can take several minutes. This can be revised when the performances improve.
  110.             while (continueCounter.getCount() > 0) {
  111.                 batch.forEach(callback);
  112.                 res = runCommand({getMore: cursorId, collection: collName});
  113.                 batch = res.cursor.nextBatch;
  114.             }
  115.             return state.value;
  116.         } catch (e) {
  117.             print("ChangeStreamListenerThread interrupted by error");
  118.             // When the error is raised by assert.commandWorked then it is already logged so we
  119.             // don't print all the details here.
  120.             if (!("codeName" in e)) {
  121.                 print(tojson(e));
  122.             }
  123.             return {"error": true, "message": tojson(e)};
  124.         }
  125.     }
  127.     /**
  128.      * Returns true if the listener thread has been created and started.
  129.      */
  130.     this.hasStarted = function() {
  131.         // Double ! to return true if '_thread' is truthy or false if it is falsy.
  132.         return !!_thread;
  133.     };
  135.     this.isReady = function() {
  136.         return _notReadyCounter.getCount() === 0;
  137.     };
  139.     /**
  140.      * Spawns a ScopedThread that will run _readChangeStream().
  141.      */
  142.     this.start = function() {
  143.         if (_thread) {
  144.             throw new Error("Listener thread is already active.");
  145.         }
  147.         _continueCounter = CountDownLatch(1);
  148.         _notReadyCounter = CountDownLatch(1);
  149.         _thread = new ScopedThread(_readChangeStream, _continueCounter, _notReadyCounter,
  150.                                    dbName, collName, _filterPipeline,
  151.                                    _changeStreamOptions, _changeHandler, _changeHandlerOptions,
  152.                                    targetHost, username, password, authEnabled, readPreferenceMode);
  153.         _thread.start();
  154.     };
  156.     /** Stops the thread. */
  157.     this.stop = function() {
  158.         if (!_thread) {
  159.             throw new Error("Listener thread is not active.");
  160.         }
  162.         _continueCounter.countDown();
  163.         _continueCounter = null;
  164.     };
  166.     /** Joins the thread. */
  167.     this.join = function() {
  168.         if (!_thread) {
  169.             throw new Error("Listener thread is not active.");
  170.         }
  171.         if (_continueCounter) {
  172.             throw new Error("Listener thread has not been stopped");
  173.         }
  174.         _thread.join();
  175.         this.returnData = _thread.returnData();
  176.         _thread = null;
  177.     };
  178. };
  181. /**
  182.  * Starts change stream listeners.
  183.  *
  184.  * @param {Number} nbListeners - The number of change stream listeners to start.
  185.  * @param {String} dbName - The database name.
  186.  * @param {String} collName - The collection name.
  187.  * @param {Array} filterPipeline - The aggregation pipeline stages to append to the '$changeStream'
  188.  *        stage.
  189.  * @param {Object} changeStreamOptions - The options to pass to the '$changeStream' stage.
  190.  * @param {Function} changeHandler - A callback function that takes a change document and a state
  191.  *        object as parameters. The function will be called for each change document read.
  192.  * @param {Object} changeHandlerOptions - An object that will be passed as argument to the
  193.  *        changeHandler callback function. Defaults to an empty object.
  194.  * @param {Array} mongoHosts - An array of mongos hosts to connect to. The listeners are assigned to
  195.  *        each host sequentially (round robin). Optional.
  196.  * @param {String} readPreferenceMode - The read preference of the change stream aggregation pipeline.
  197.  * @return {Array} The created listeners.
  198.  */
  199. function startListeners(nbListeners, dbName, collName, filterPipeline,
  200.                         changeStreamOptions, changeHandler, changeHandlerOptions, mongosHosts, targetHost, readPreferenceMode)
  201. {
  202.     if (mongosHosts && mongosHosts.length !== 0) {
  203.         print("Starting " + nbListeners + " threads split on targets: " + tojson(mongosHosts));
  204.     } else {
  205.         print("Starting " + nbListeners + " threads");
  206.     }
  207.     var listeners = [];
  208.     for (var i = 0; i < nbListeners; i++) {
  209.         if (mongosHosts) {
  210.             targetHost = mongosHosts[i % mongosHosts.length];
  211.         }
  212.         listener = new ChangeStreamListenerThread(dbName, collName, filterPipeline,
  213.                                                   changeStreamOptions, changeHandler,
  214.                                                   changeHandlerOptions, targetHost, readPreferenceMode);
  215.         listener.start();
  216.         listeners.push(listener);
  217.     }
  218.     return listeners;
  219. }
  222. /**
  223.  * Waits for the listeners to be ready.
  224.  *
  225.  * @param {Array} listeners - The listeners as an array of ChangeStreamListenerThreads.
  226.  */
  227. function waitForListeners(listeners) {
  228.     var start = new Date().getTime();
  229.     // Timeout is 5s + 100ms * nb of listeners.
  230.     // It is an arbitrary value aimed at giving enough time for all the listeners to start so
  231.     // that when we hit a timeout it is clear something went wrong.
  232.     var end_timeout = start + 5000 + listeners.length * 100;
  233.     do {
  234.         var allReady = true;
  235.         for (var i = 0; i < listeners.length; i++) {
  236.             if (!listeners[i].isReady()) {
  237.                 allReady = false;
  238.                 sleep(50);
  239.                 break;
  240.             }
  241.         }
  242.         if (allReady) {
  243.             return;
  244.         }
  245.     } while (new Date().getTime() < end_timeout);
  246.     throw new Error("Timeout while waiting to for change stream listeners to be ready.");
  247. }
  250. /**
  251.  * Starts and waits for the change stream listeners to be ready.
  252.  *
  253.  * @param {String} dbName - The database name.
  254.  * @param {Array} collNames - An array of collection names on which to listen for change streams.
  255.  * @param {Number} nListeners - The number of change streams listeners to start. Should be a
  256.  *        multiple of the number of collections.
  257.  * @param {Object} changeStreamOptions - Options for the '$changeStream' stage.
  258.  * @param {Boolean} selectiveChange - Whether additional aggregation stages should be added
  259.  *        after the '$changeStream' stage.
  260.  * @return {Object} A dictionary mapping collection names to arrays of corresponding listeners.
  261.  */
  262. function startAndWaitForListeners(dbName, collNames, nListeners, changeStreamOptions, selectiveChange,
  263.                                   changeHandler, changeHandlerOptions, mongosHosts, targetHost, readPreferenceMode) {
  264.     print("Starting the change stream listeners");
  265.     var listeners = {};
  266.     var collName = null;
  267.     var filterPipeline = getFilterPipeline(selectiveChange);
  268.     var nCollListeners = nListeners / collNames.length;
  270.     for (var i = 0; i < collNames.length; i++) {
  271.         collName = collNames[i];
  272.         listeners[collName] = startListeners(nCollListeners, dbName, collName, filterPipeline,
  273.                                              changeStreamOptions, changeHandler, changeHandlerOptions,
  274.                                              mongosHosts, targetHost, readPreferenceMode);
  275.     }
  276.     print("The listeners are started");
  278.     print("Waiting for the listeners to be ready");
  279.     for (collName in listeners) {
  280.         waitForListeners(listeners[collName]);
  281.     }
  282.     print("All the listeners are ready");
  283.     return listeners;
  284. }
  286. /**
  287.  * Stops the change stream listeners.
  288.  *
  289.  * @param {Array} listeners - An array of ChangeStreamListenerThread objects.
  290.  * @return {Array} An array containing the final state values for all the listeners.
  291.  */
  292. function stopListeners(listeners) {
  293.     var states = [];
  294.     var listener;
  295.     for (var i = 0; i < listeners.length; i++) {
  296.         listeners[i].stop();
  297.     }
  298.     for (var j = 0; j < listeners.length; j++) {
  299.         listener = listeners[j];
  300.         listener.join();
  301.         if (listener.returnData.error) {
  302.             throw new Error("A change stream listener failed with error: '" +
  303.                             listener.returnData.message + "'");
  304.         }
  305.         states.push(listener.returnData);
  306.     }
  307.     return states;
  308. }
  311. /**********************************************************************
  312.  * Helper functions to handle the change streams tests configuration. *
  313.  **********************************************************************/
  315. /**
  316.  * Transforms a mongos host list as found infrastructure.out.yml into
  317.  * a list of host strings that can be used to create a new Mongo connection.
  318.  */
  319. function getMongosHosts(mongosHostsConfig) {
  320.     var mongos_targets = [];
  321.     for (var i = 0; i < mongosHostsConfig.length; i++) {
  322.         mongos_targets.push(mongosHostsConfig[i].private_ip);
  323.     }
  324.     return mongos_targets;
  325. }
  328. /**
  329.  * Generates and returns a list of collection names.
  330.  *
  331.  * @param {String} baseName - The collections base name.
  332.  * @param {Number} nCollections - The number of collection names to generate.
  333.  * @return {Array} An array of collection names.
  334.  */
  335. function generateCollectionNames(baseName, nCollections) {
  336.     var names = [];
  337.     for (var i = 0; i < nCollections; i++) {
  338.         names.push(baseName + "_" + i);
  339.     }
  340.     return names;
  341. }
  344. /**
  345.  * Gets the filter pipeline that should follow the '$changeStream' aggregation stage.
  346.  *
  347.  * @param {bool} selectiveChange - The value of the 'selective_change' option for the test.
  348.  * @return {Array} The filter pipeline to add to the change stream pipeline.
  349.  */
  350. function getFilterPipeline(selectiveChange) {
  351.     if (selectiveChange) {
  352.         return [{"$match": {"a": {"$lt": 100000}}}];
  353.     } else {
  354.         return [];
  355.     }
  356. }
  358. /**
  359.  * Indicates if the servers use a resume token with the BinData format.
  360.  * This format is used before 3.7.4 and also when featureCompatibilityVersion is 3.6
  361.  * (change streams are only supported since 3.6).
  362.  */
  363. function useBinDataResumeToken() {
  364.     var version = db.version().split(".");
  365.     if (version < [3, 7, 4]) {
  366.         return true;
  367.     }
  368.     if (!sharded_cluster()) {
  369.         // Mongos instances do not give access to the featureCompatibilityVersion.
  370.         // On a sharded cluster we currently assume the FCV is not set to 3.6.
  371.         // PERF-1447 was filed to handle FCV on sharded clusters in the future.
  372.         fcvRes = db.adminCommand({ getParameter: 1, featureCompatibilityVersion: 1});
  373.         fcv = fcvRes.featureCompatibilityVersion.version;
  374.         if (fcv == "3.6") {
  375.             return true;
  376.         }
  377.     }
  378.     return false;
  379. }
  382. /**********************************************************************
  383.  * Other common functions shared by the change stream test workloads. *
  384.  **********************************************************************/
  386. /**
  387.  * Gets the base name for a test given its configuration options.
  388.  *
  389.  * @param {Number} nListeners - The number of change stream listeners.
  390.  * @param {String} fullDocumentBeforeChange - The "fullDocumentBeforeChange" change stream option.
  391.  * @param {String} fullDocument - The "fullDocument" change stream option.
  392.  * @param {Boolean} selectiveChange - Whether the 'selective_change' option is set.
  393.  * @param {Number} nCollections - The number of collections used by the test.
  394.  * @param {Number} docSize - The size of the documents inserted into the collection.
  395.  * @param {Boolean} preImageEnabled - Whether recording of the pre-images is enabled.
  396.  * @param {String} readPreferenceMode - The read preference of the change stream aggregation pipeline.
  397.  * @return {String} The test base name.
  398.  */
  399. function getTestBaseName(nListeners, fullDocumentBeforeChange, fullDocument, selectiveChange, nCollections, docSize, preImageEnabled, readPreferenceMode) {
  400.     var baseName = `${nListeners}`;
  401.     if (fullDocumentBeforeChange) {
  402.         baseName += `_${fullDocumentBeforeChange}fdbc`;
  403.     }
  404.     if (fullDocument) {
  405.         if (fullDocument == "updateLookup") {
  406.             // Previously, change stream tests that had fullDocument being set to "updateLookup",
  407.             // were having a prefix of "_lookup". The newly added prefix is omitted for this
  408.             // fullDocument parameter in order to preserve the historical data.
  409.             baseName += "_lookup";
  410.         } else {
  411.             baseName += `_${fullDocument}fd`;
  412.         }
  413.     }
  414.     if (selectiveChange) {
  415.         baseName += "_filter";
  416.     }
  417.     if (preImageEnabled) {
  418.         baseName += "_preImage";
  419.     }
  420.     baseName += `_${nCollections}c`;
  422.     // Previously, change stream tests were only executed with document size of 100 bytes. The newly
  423.     // added prefix is omitted for this document size in order to preserve the historical data.
  424.     if (docSize != 100) {
  425.         baseName += `_${docSize}d`;
  426.     }
  428.     if (readPreferenceMode != "primary") {
  429.         baseName += `_${readPreferenceMode}`;
  430.     }
  432.     return baseName;
  433. }
  435. /**
  436.  * Returns the change stream options object for the specified configuration.
  437.  *
  438.  * @param {String} fullDocumentBeforeChange - The 'fullDocumentBeforeChange' parameter of the $changeStream stage.
  439.  * @param {String} fullDocument - The 'fullDocument' parameter of the $changeStream stage.
  440.  */
  441. function getChangeStreamOptions(fullDocumentBeforeChange, fullDocument) {
  442.     let result = {};
  443.     if (fullDocumentBeforeChange != undefined) {
  444.         result["fullDocumentBeforeChange"] = fullDocumentBeforeChange;
  445.     }
  446.     if (fullDocument != undefined) {
  447.         result["fullDocument"] = fullDocument;
  448.     }
  449.     return result;
  450. }
  452. /**
  453.   * Returns true if the FCV is 6.1 or above (when the feature flag is removed) or if feature flag
  454.   * 'featureFlagClusterWideConfig' is enabled with FCV 6.0, false otherwise.
  455.   */
  456. function isClusterWideConfigFeatureAvailable(db) {
  457.     const getParam = db.adminCommand({getParameter: 1, featureFlagClusterWideConfig: 1});
  458.     // On replica sets, the FCV needs to be checked so that the last LTS/continuous FCV replset variants
  459.     // are accounted for. On sharded clusters, the FCV cannot be retrieved directly from the mongos
  460.     // so we use the binary version instead.
  461.     if (!sharded_cluster()) {
  462.         const fcv = db.adminCommand({getParameter: 1, featureCompatibilityVersion: 1});
  463.         const majorFCV = fcv.featureCompatibilityVersion.version.split('.')[0];
  464.         const minorFCV = fcv.featureCompatibilityVersion.version.split('.')[1];
  465.         return (majorFCV > 6 || (majorFCV == 6 && minorFCV >= 1)) || (getParam.hasOwnProperty("featureFlagClusterWideConfig") &&
  466.                                     (getParam.featureFlagClusterWideConfig.value) && (majorFCV == 6) && (minorFCV == 0));
  467.     } else {
  468.         const majorVersion = db.version().split('.')[0];
  469.         const minorVersion = db.version().split('.')[1];
  470.         return (majorVersion > 6 || (majorVersion == 6 && minorVersion >= 1)) || (getParam.hasOwnProperty("featureFlagClusterWideConfig") &&
  471.                                     getParam.featureFlagClusterWideConfig.value);
  472.     }
  473. }
  475. function testChangeStreamsCRUDWorkload(db,
  476.                                        testName,
  477.                                        { dbName, collNames, docSize, nCollections, nThreads, nListeners, nDocs, retryableWrites, mongosHosts, shardCollections, preImagesEnabled, changeStreamOptions, fullDocumentBeforeChange, fullDocument, selectiveChange, changeHandlerOptions, testDurationSecs, changeStreamReadPreference },
  478.                                        targetHost,
  479.                                        changeHandler,
  480.                                        onStopCallback,
  481.                                        writeOnlyWorkload) {
  482.     // Set the number of documents to be inserted before running the workload.
  483.     var nDocuments = nDocs != undefined ? nDocs : 10000000 / docSize;
  484.     var workload = new MixedCRUDWorkload(dbName, collNames, nThreads, retryableWrites,
  485.                                          mongosHosts, docSize, nDocuments, shardCollections, writeOnlyWorkload);
  487.     // Note: the workload initialization calls quiesce at the end.
  488.     workload.initialize();
  490.     if (changeStreamOptions && !isClusterWideConfigFeatureAvailable(db)) {
  491.       jsTestLog("This test requires the time-based change stream pre-/post-image retention policy to be available. Skipping test.");
  492.       return;
  493.     }
  494.     if (preImagesEnabled) {
  495.       for (var i = 0; i < collNames.length; i++) {
  496.         assert.commandWorked(db.getSiblingDB(dbName).runCommand({collMod: collNames[i], changeStreamPreAndPostImages: {enabled: true}}));
  497.       }
  498.     }
  499.     if (isClusterWideConfigFeatureAvailable(db)) {
  500.       var changeStreamOptions = changeStreamOptions ? changeStreamOptions : {preAndPostImages: {expireAfterSeconds: "off"}};
  501.       assert.commandWorked(db.getSiblingDB("admin").runCommand(Object.merge({setClusterParameter: {changeStreamOptions: changeStreamOptions}})));
  502.     }
  504.     jsTest.log("Change streams CRUD " + testName + ": nThreads=" + nThreads + " nListeners=" + nListeners + " preImagesEnabled=" + preImagesEnabled + " docSize=" + docSize + " nCollections=" + nCollections + " shardCollections=" + shardCollections + " " + new Date());
  506.     var changeStreamOptions = getChangeStreamOptions(fullDocumentBeforeChange, fullDocument);
  507.     var listeners = startAndWaitForListeners(dbName, collNames, nListeners, changeStreamOptions,
  508.                                              selectiveChange, changeHandler,
  509.                                              changeHandlerOptions, mongosHosts, targetHost, changeStreamReadPreference);
  511.     print("Starting the workload");
  512.     workload.start();
  513.     sleep(testDurationSecs*1000);
  514.     print("Finishing the workload");
  515.     var throughput = workload.stop();
  517.     onStopCallback(throughput, listeners, getTestBaseName(nListeners, fullDocumentBeforeChange, fullDocument, selectiveChange, nCollections, docSize, preImagesEnabled, changeStreamReadPreference));
  518. }