path: root/jstests/noPassthrough/set_step_params.js

load("jstests/libs/parallelTester.js");

/**
 * @tags: [requires_replication, requires_sharding]
 */

(function() {
    "use strict";

    const kDbName = 'test';

    const minConns = 4;
    var stepParams = {
        ShardingTaskExecutorPoolMinSize: minConns,
        ShardingTaskExecutorPoolMaxSize: 10,
        ShardingTaskExecutorPoolMaxConnecting: 5,
        ShardingTaskExecutorPoolHostTimeoutMS: 300000,
        ShardingTaskExecutorPoolRefreshRequirementMS: 60000,
        ShardingTaskExecutorPoolRefreshTimeoutMS: 20000,
        ShardingTaskExecutorPoolReplicaSetMatching: "disabled",
    };

    const st = new ShardingTest({
        config: {nodes: 1},
        shards: 1,
        rs0: {nodes: 1},
        mongos: [{setParameter: stepParams}],
    });
    const mongos = st.s0;
    const rst = st.rs0;
    const primary = rst.getPrimary();

    const cfg = primary.getDB('local').system.replset.findOne();
    const allHosts = cfg.members.map(x => x.host);
    const mongosDB = mongos.getDB(kDbName);
    const primaryOnly = [primary.name];

    function configureReplSetFailpoint(name, modeValue) {
        st.rs0.nodes.forEach(function(node) {
            assert.commandWorked(node.getDB("admin").runCommand({
                configureFailPoint: name,
                mode: modeValue,
                data: {shouldCheckForInterrupt: true},
            }));
        });
    }

    var threads = [];
    function launchFinds({times, readPref, shouldFail}) {
        jsTestLog("Starting " + times + " connections");
        for (var i = 0; i < times; i++) {
            var thread = new Thread(function(connStr, readPref, dbName, shouldFail) {
                var client = new Mongo(connStr);
                const ret = client.getDB(dbName).runCommand(
                    {find: "test", limit: 1, "$readPreference": {mode: readPref}});

                if (shouldFail) {
                    assert.commandFailed(ret);
                } else {
                    assert.commandWorked(ret);
                }
            }, st.s.host, readPref, kDbName, shouldFail);
            thread.start();
            threads.push(thread);
        }
    }

    var currentCheckNum = 0;
    function hasConnPoolStats(args) {
        const checkNum = currentCheckNum++;
        jsTestLog("Check #" + checkNum + ": " + tojson(args));
        var {ready, pending, active, hosts, isAbsent} = args;

        ready = ready ? ready : 0;
        pending = pending ? pending : 0;
        active = active ? active : 0;
        hosts = hosts ? hosts : allHosts;

        function checkStats(res, host) {
            var stats = res.hosts[host];
            if (!stats) {
                jsTestLog("Connection stats for " + host + " are absent");
                return isAbsent;
            }

            jsTestLog("Connection stats for " + host + ": " + tojson(stats));
            return stats.available == ready && stats.refreshing == pending && stats.inUse == active;
        }

        function checkAllStats() {
            var res = mongos.adminCommand({connPoolStats: 1});
            return hosts.map(host => checkStats(res, host)).every(x => x);
        }

        assert.soon(checkAllStats, "Check #" + checkNum + " failed", 10000);

        jsTestLog("Check #" + checkNum + " successful");
    }

    function updateSetParameters(params) {
        var cmd = Object.assign({"setParameter": 1}, params);
        assert.commandWorked(mongos.adminCommand(cmd));
    }

    function dropConnections() {
        assert.commandWorked(mongos.adminCommand({dropConnections: 1, hostAndPort: allHosts}));
    }

    function resetPools() {
        dropConnections();
        mongos.adminCommand({multicast: {ping: 0}});
        hasConnPoolStats({ready: 4});
    }

    function runSubTest(name, fun) {
        jsTestLog("Running test for " + name);

        resetPools();

        fun();

        updateSetParameters(stepParams);
    }

    assert.writeOK(mongosDB.test.insert({x: 1}));
    assert.writeOK(mongosDB.test.insert({x: 2}));
    assert.writeOK(mongosDB.test.insert({x: 3}));
    st.rs0.awaitReplication();

    runSubTest("MinSize", function() {
        dropConnections();

        // Launch an initial find to trigger to min
        launchFinds({times: 1, readPref: "primary"});
        hasConnPoolStats({ready: minConns});

        // Increase by one
        updateSetParameters({ShardingTaskExecutorPoolMinSize: 5});
        hasConnPoolStats({ready: 5});

        // Increase to MaxSize
        updateSetParameters({ShardingTaskExecutorPoolMinSize: 10});
        hasConnPoolStats({ready: 10});

        // Decrease to zero
        updateSetParameters({ShardingTaskExecutorPoolMinSize: 0});
    });

    runSubTest("MaxSize", function() {
        configureReplSetFailpoint("waitInFindBeforeMakingBatch", "alwaysOn");
        dropConnections();

        // Launch 10 blocked finds
        launchFinds({times: 10, readPref: "primary"});
        hasConnPoolStats({active: 10, hosts: primaryOnly});

        // Increase by 5 and Launch another 4 blocked finds
        updateSetParameters({ShardingTaskExecutorPoolMaxSize: 15});
        launchFinds({times: 4, readPref: "primary"});
        hasConnPoolStats({active: 14, hosts: primaryOnly});

        // Launch yet another 2, these should add only 1 connection
        launchFinds({times: 2, readPref: "primary"});
        hasConnPoolStats({active: 15, hosts: primaryOnly});

        configureReplSetFailpoint("waitInFindBeforeMakingBatch", "off");
        hasConnPoolStats({ready: 15, pending: 0, hosts: primaryOnly});
    });

    // Test maxConnecting
    runSubTest("MaxConnecting", function() {
        const maxPending1 = 2;
        const maxPending2 = 4;
        const conns = 6;

        updateSetParameters({
            ShardingTaskExecutorPoolMaxSize: 100,
            ShardingTaskExecutorPoolMaxConnecting: maxPending1,
        });

        configureReplSetFailpoint("waitInIsMaster", "alwaysOn");
        configureReplSetFailpoint("waitInFindBeforeMakingBatch", "alwaysOn");
        dropConnections();

        // Go to the limit of maxConnecting, so we're stuck here
        launchFinds({times: maxPending1, readPref: "primary"});
        hasConnPoolStats({pending: maxPending1});

        // More won't run right now
        launchFinds({times: conns - maxPending1, readPref: "primary"});
        hasConnPoolStats({pending: maxPending1});

        // If we increase our limit, it should fill in some of the connections
        updateSetParameters({ShardingTaskExecutorPoolMaxConnecting: maxPending2});
        hasConnPoolStats({pending: maxPending2});

        // Dropping the limit doesn't cause us to drop pending
        updateSetParameters({ShardingTaskExecutorPoolMaxConnecting: maxPending1});
        hasConnPoolStats({pending: maxPending2});

        // Release our pending and walk away
        configureReplSetFailpoint("waitInIsMaster", "off");
        hasConnPoolStats({active: conns});
        configureReplSetFailpoint("waitInFindBeforeMakingBatch", "off");
    });

    runSubTest("Timeouts", function() {
        const conns = minConns;
        const pendingTimeoutMS = 5000;
        const toRefreshTimeoutMS = 1000;
        const idleTimeoutMS1 = 20000;
        const idleTimeoutMS2 = 15500;

        // Updating separately since the validation depends on existing params
        updateSetParameters({
            ShardingTaskExecutorPoolRefreshTimeoutMS: pendingTimeoutMS,
        });
        updateSetParameters({
            ShardingTaskExecutorPoolRefreshRequirementMS: toRefreshTimeoutMS,
        });
        updateSetParameters({
            ShardingTaskExecutorPoolHostTimeoutMS: idleTimeoutMS1,
        });

        configureReplSetFailpoint("waitInFindBeforeMakingBatch", "alwaysOn");
        dropConnections();

        // Make ready connections
        launchFinds({times: conns, readPref: "primary"});
        configureReplSetFailpoint("waitInFindBeforeMakingBatch", "off");
        hasConnPoolStats({ready: conns});

        // Block refreshes and wait for the toRefresh timeout
        configureReplSetFailpoint("waitInIsMaster", "alwaysOn");
        sleep(toRefreshTimeoutMS);

        // Confirm that we're in pending for all of our conns
        hasConnPoolStats({pending: conns});

        // Set our min conns to 0 to make sure we don't refresh after pending timeout
        updateSetParameters({
            ShardingTaskExecutorPoolMinSize: 0,
        });

        // Wait for our pending timeout
        sleep(pendingTimeoutMS);
        hasConnPoolStats({});

        configureReplSetFailpoint("waitInIsMaster", "off");

        // Reset the min conns to make sure normal refresh doesn't extend the timeout
        updateSetParameters({
            ShardingTaskExecutorPoolMinSize: minConns,
        });

        // Wait for our host timeout and confirm the pool drops
        sleep(idleTimeoutMS1);
        hasConnPoolStats({isAbsent: true});

        // Reset the pool
        resetPools();

        // Sleep for a shorter timeout and then update so we're already expired
        sleep(idleTimeoutMS2);
        updateSetParameters({ShardingTaskExecutorPoolHostTimeoutMS: idleTimeoutMS2});
        hasConnPoolStats({isAbsent: true});
    });

    threads.forEach(function(thread) {
        thread.join();
    });

    st.stop();
})();