locust_workload.py

#!/usr/bin/env python3
#
# Locust-based create/read/update workload, based on the test documents generated by the mgodatagen
# configs (locust_workload_mgodatagen_*.json).
#
# The shard key range and namespace are inferred from the mgodatagen JSON config file.
#
# Example usage:
#   # First, populate the collection:
#   mgodatagen --seed 1777546439 -f locust_workload_mgodatagen_100GB.json --uri mongodb://localhost
#   mgodatagen --seed 1777546439 -f locust_workload_mgodatagen_100GB_autoinc.json --uri mongodb://localhost
#
#   # For collection population against a standalone server (faster), use the following command to start MongoD in standalone mode:
#   ./mongod --dbpath /mnt/data/mongod --logpath /mnt/data/mongod/mongod.log --wiredTigerCacheSizeGB 18 --port 27017 --bind_ip_all --fork
#   nohup mgodatagen --seed 1777546439 -f locust_workload_mgodatagen_1TB.json --uri mongodb://localhost 2>&1 &
#
#   # Then, run the workload:
#   locust -f locust_workload.py --csv=locust_results --csv-full-history --html=locust_results_report.html --print-stats --mgodatagen-config locust_workload_mgodatagen_100GB.json --host mongodb://localhost
#   locust -f locust_workload.py --csv=locust_results --csv-full-history --html=locust_results_report.html --print-stats --mgodatagen-config locust_workload_mgodatagen_100GB_autoinc.json --host mongodb://localhost
#
#   # Unattended workload run on AWS hosts
#   ./launch_ec2_replicaset_hosts.py <Run Name> launch
#   ./remote_control_replicaset.py <Run Name> deploy-binaries
#   ./remote_control_replicaset.py <Run Name> init
#
#   nohup locust -f locust_workload.py --csv=locust_results --csv-full-history --html=locust_results_report.html --print-stats --mgodatagen-config locust_workload_mgodatagen_1TB.json --auto-execute [deleteMany_10_pct, fastBulkDelete_10_pct] --host mongodb:// > locust_results_nohup.log 2>&1 &
#
#   # Curl commands
#   curl -sX POST http://localhost:8090/custom_actions/deleteMany_10_pct | jq -r .command
#   curl -sX POST http://localhost:8090/custom_actions/fastBulkDelete_10_pct | jq -r .command
#
#   # Add ?execute=true to actually run the command in the background (one instance at a time):
#   curl -sX POST "http://localhost:8090/custom_actions/deleteMany_10_pct?execute=true"
#   curl -sX POST "http://localhost:8090/custom_actions/fastBulkDelete_10_pct?execute=true"
#
#   # warm_up is read-only and runs immediately in the background (no execute flag).
#   curl -sX POST http://localhost:8090/custom_actions/warm_up
#
#   # Pass --warm-up at startup to automatically scan all secondary indexes once locust initializes.
#
#   # Download a visual (html) report from a running instance
#   curl -L -o locust_results_report.html "http://localhost:8090/stats/report?download=1&theme=dark"
#
#   # Download the full request statistics history (csv) from a running instance
#   curl -L -o locust_results_stats_history.csv "http://localhost:8090/stats/requests_full_history/csv?download=1"
#

import json
import logging
import threading

import locust.stats

from flask import jsonify, render_template_string, request as flask_request
from locust import User, constant_pacing, events, task
from pymongo import MongoClient, ReadPreference
from random import choice, randint, uniform
from time import perf_counter_ns

# Percentiles to capture in the UI response times graph
locust.stats.PERCENTILES_TO_CHART = [0.5, 0.90, 0.99]

# Percentiles to capture in the UI statistics tab
locust.stats.PERCENTILES_TO_STATISTICS = [0.90, 0.99]

# Percentiles to capture in the statistics history (.csv files)
locust.stats.PERCENTILES_TO_REPORT = [0.50, 0.75, 0.90, 0.99, 0.999]

# Make the request name column much narrower for the terminal/headless mode
locust.stats.STATS_TYPE_WIDTH = 1
locust.stats.STATS_NAME_WIDTH = 50

connection_string = None
collection = None

# Populated from mgodatagen config at init time
AVG_DOC_BYTES = 0
SECONDARY_INDEX_FIELDS = []
SHARD_KEY_IS_STRING = True
SHARD_KEY_LENGTH = 0
SHARD_KEY_RANGE_START = 0
SHARD_KEY_RANGE_END = 0
COLLECTION_COUNT = 0

# Hardcoded, not populated from mgodatagen config
SECONDARY_KEY_LENGTH = 120


def compute_avg_doc_bytes(content):
    """
    Estimate average BSON document size from a mgodatagen content spec.
    """
    # BSON document: 4-byte size prefix + fields + 1-byte null terminator
    total = 5
    # _id is always present as ObjectId (12 bytes): type + "_id" + null + value
    total += 1 + 3 + 1 + 12
    for field_name, field_spec in content.items():
        field_type = field_spec.get('type', '')
        # Each element: type byte + field name + null terminator
        header = 1 + len(field_name) + 1
        if field_type == 'string':
            avg_len = (field_spec.get('minLength', 0) + field_spec.get('maxLength', 0)) / 2
            # BSON string: 4-byte length int32 + content bytes + null terminator
            total += header + 4 + avg_len + 1
        elif field_type == 'autoincrement':
            # BSON int64: 8-byte little-endian signed integer
            total += header + 8
        else:
            raise ValueError(f'Unsupported field type for {field_name}: {field_spec}')
    return round(total)


# Obtained from https://github.com/feliixx/mgodatagen#string
MGODATAGEN_STRING_CHARSET = 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-_'


def random_shard_key():
    if SHARD_KEY_IS_STRING:
        return ''.join(choice(MGODATAGEN_STRING_CHARSET) for _ in range(SHARD_KEY_LENGTH))
    else:
        return randint(SHARD_KEY_RANGE_START, SHARD_KEY_RANGE_END - 1)


def random_secondary_key():
    return ''.join(choice(MGODATAGEN_STRING_CHARSET) for _ in range(SECONDARY_KEY_LENGTH))


def nanos_to_millis(nanos):
    return round(nanos / 1000000.0, 2)


# Cool-down after the auto-execute action completes before the runner quits, so final stats can flush.
_AUTO_EXECUTE_QUIT_DELAY_SECS = 3600


@events.init_command_line_parser.add_listener
def on_locust_init_command_line_parser(parser):
    # Locust arguments
    parser.set_defaults(web_port=8090, num_users=1000, spawn_rate=10, autostart=True)

    # Custom arguments
    parser.add_argument('--mgodatagen-config', help='Path to the mgodatagen JSON config file',
                        metavar='config', type=str, required=True)
    # The choices must match the /custom_actions/* routes below
    parser.add_argument(
        '--auto-execute',
        choices=('deleteMany_10_pct', 'fastBulkDelete_10_pct'),
        default=None,
        help='Automatically execute the specified action after --auto-execute-delay seconds',
    )
    parser.add_argument(
        '--auto-execute-delay',
        type=int,
        default=3600,
        metavar='SECONDS',
        help='Warm-up delay before auto-execute fires (default: 3600s)',
    )
    parser.add_argument(
        '--warm-up',
        action='store_true',
        default=False,
        help='At startup, scan every secondary index to bring it into the WiredTiger cache',
    )


@events.init.add_listener
def on_locust_init(environment, **kwargs):
    global connection_string
    connection_string = 'mongodb://localhost' if environment.host is None else environment.host

    # Load the mgodatagen config to infer namespace and shard key range
    with open(environment.parsed_options.mgodatagen_config) as f:
        config = json.load(f)[0]

    ns_db = config['database']
    ns_coll = config['collection']

    global COLLECTION_COUNT
    COLLECTION_COUNT = config['count']

    global SHARD_KEY_IS_STRING, SHARD_KEY_LENGTH, SHARD_KEY_RANGE_START, SHARD_KEY_RANGE_END
    shard_key_spec = config['content']['shardKey']
    shard_key_type = shard_key_spec['type']
    if shard_key_type == 'string':
        SHARD_KEY_IS_STRING = True
        SHARD_KEY_LENGTH = shard_key_spec['maxLength']
    elif shard_key_type == 'autoincrement' and shard_key_spec.get('autoType') == 'long':
        SHARD_KEY_IS_STRING = False
        SHARD_KEY_RANGE_START = shard_key_spec.get('startLong', 0)
        SHARD_KEY_RANGE_END = SHARD_KEY_RANGE_START + COLLECTION_COUNT
    else:
        raise ValueError(f'Unsupported shardKey type: {shard_key_spec}')

    global AVG_DOC_BYTES
    AVG_DOC_BYTES = compute_avg_doc_bytes(config['content'])

    # Extract secondary index fields (exclude the shardKey index)
    global SECONDARY_INDEX_FIELDS
    for idx in config.get('indexes', []):
        field = list(idx['key'].keys())[0]
        if field != 'shardKey':
            SECONDARY_INDEX_FIELDS.append(field)

    logging.info(f'MongoDB host: {connection_string}')
    logging.info(f'Namespace: {ns_db}.{ns_coll}')
    if SHARD_KEY_IS_STRING:
        logging.info(f'Shard key type: string (length {SHARD_KEY_LENGTH})')
    else:
        logging.info(f'Shard key type: autoincrement '
                     f'(range [{SHARD_KEY_RANGE_START}, {SHARD_KEY_RANGE_END}))')
    logging.info(f'Secondary key length: {SECONDARY_KEY_LENGTH}')
    logging.info(f'Avg document size: {AVG_DOC_BYTES} bytes')
    logging.info(f'Secondary index fields: {SECONDARY_INDEX_FIELDS}')
    if environment.parsed_options.auto_execute:
        logging.info(f'Auto-execute action: {environment.parsed_options.auto_execute}')
        logging.info(f'Auto-execute delay: {environment.parsed_options.auto_execute_delay}s')

    global collection
    mongo_client = MongoClient(connection_string, connectTimeoutMS=30000, maxPoolSize=150)
    collection = mongo_client[ns_db].get_collection(ns_coll, read_preference=ReadPreference.PRIMARY)

    if environment.web_ui:
        # http://host:8089/custom_actions
        _register_custom_actions(environment.web_ui.app, environment)


class MongoUser(User):
    '''
    This user will generate a constant load of 1 request per second.
    '''

    wait_time = constant_pacing(1)

    def on_start(self):
        self.shard_key = None
        self.select_shard_key()

    @task(50)
    def select_shard_key(self):
        # Probe a random point in the key space and find the first document at or after it.  This
        # exercises the shardKey index without requiring an exact match and produces a uniformly
        # distributed random document selection regardless of the actual key distribution in the
        # collection.
        probe = random_shard_key()

        start_time = perf_counter_ns()

        doc = collection.with_options(read_preference=ReadPreference.SECONDARY_PREFERRED).find_one(
            filter={'shardKey': {
                '$gte': probe
            }},
            sort=[('shardKey', 1)],
        )

        if doc is None:
            # Probe landed beyond the last key; wrap around to the first document.
            doc = collection.with_options(
                read_preference=ReadPreference.SECONDARY_PREFERRED).find_one(
                    {},
                    sort=[('shardKey', 1)],
                )

        elapsed = perf_counter_ns() - start_time

        if doc is not None:
            self.shard_key = doc['shardKey']

        self.environment.events.request.fire(
            request_type='r',
            name='select_shard_key',
            response_time=nanos_to_millis(elapsed),
            response_length=0,
            exception=None,
        )

    @task(25)
    def update_by_shard_key(self):
        if self.shard_key is None:
            self.select_shard_key()
            return

        start_time = perf_counter_ns()

        update_result = collection.update_one(
            filter={'shardKey': self.shard_key},
            update={'$inc': {
                'updates': 1
            }},
            upsert=False,
        )

        if update_result.modified_count == 0:
            self.select_shard_key()
            return

        elapsed = perf_counter_ns() - start_time

        self.environment.events.request.fire(
            request_type='w',
            name='update_by_shard_key',
            response_time=nanos_to_millis(elapsed),
            response_length=0,
            exception=None,
        )

    @task(20)
    def select_shard_key_by_secondary_index(self):
        if not SECONDARY_INDEX_FIELDS:
            return

        field = choice(SECONDARY_INDEX_FIELDS)
        probe = random_secondary_key()

        start_time = perf_counter_ns()

        doc = collection.with_options(read_preference=ReadPreference.SECONDARY_PREFERRED).find_one(
            filter={field: {
                '$gte': probe
            }},
            sort=[(field, 1)],
        )

        if doc is None:
            # Probe landed beyond the last key; wrap around to the first document.
            collection.with_options(read_preference=ReadPreference.SECONDARY_PREFERRED).find_one(
                {},
                sort=[(field, 1)],
            )

        elapsed = perf_counter_ns() - start_time

        if doc is not None:
            self.shard_key = doc['shardKey']

        self.environment.events.request.fire(
            request_type='r',
            name='select_shard_key_by_secondary_index',
            response_time=nanos_to_millis(elapsed),
            response_length=0,
            exception=None,
        )

    @task(5)
    def insert_new_shard_key(self):
        # Place the new key outside the normal key range so inserts never fall inside a shard key
        # range being deleted.
        if SHARD_KEY_IS_STRING:
            # '!' (0x21) and '~' (0x7E) sort respectively before and after every character in
            # MGODATAGEN_STRING_CHARSET ('-' 0x2D to 'z' 0x7A).
            new_key = choice(('!', '~')) + random_shard_key()
        else:
            new_key = choice((-10**15, +10**15)) + random_shard_key()

        start_time = perf_counter_ns()

        collection.update_one(
            filter={'shardKey': new_key},
            update={
                '$set': {
                    'inserted': True
                },
                '$inc': {
                    'updates': 1
                }
            },
            upsert=True,
        )

        elapsed = perf_counter_ns() - start_time

        self.environment.events.request.fire(
            request_type='w',
            name='insert_new_shard_key',
            response_time=nanos_to_millis(elapsed),
            response_length=0,
            exception=None,
        )


# ---------------------------------------------------------------------------
# Custom actions web UI (accessible at http://host:8089/custom_actions)
# ---------------------------------------------------------------------------

_CUSTOM_ACTIONS_HTML = '''<!doctype html>
<html>
<head><title>Custom Actions</title></head>
<body>
<h2>Custom Actions</h2>

<h3>deleteMany_10_pct</h3>
<p>
  Computes a shardKey range covering 10% of the collection, starting at a random
  offset between 10% and 30% from the beginning, using covered index scans.
  Returns a <code>db.runCommand</code> you can paste into mongosh.
</p>
<button onclick="runAction('/custom_actions/deleteMany_10_pct', 'out_deleteMany_10_pct')">Compute deleteMany_10_pct command</button>
<pre id="out_deleteMany_10_pct" style="margin-top:1em"></pre>

<h3>fastBulkDelete_10_pct</h3>
<p>
  Same range computation as <code>deleteMany_10_pct</code> but generates a
  <code>fastBulkDelete</code> command using <code>filterIndexName</code>,
  <code>lowerBound</code>, and <code>upperBound</code>.
</p>
<button onclick="runAction('/custom_actions/fastBulkDelete_10_pct', 'out_fastBulkDelete_10_pct')">Compute fastBulkDelete_10_pct command</button>
<pre id="out_fastBulkDelete_10_pct" style="margin-top:1em"></pre>

<h3>warm_up</h3>
<p>
  Performs a covered index scan over every secondary index in the background,
  forcing each one into the WiredTiger cache (one instance at a time).
</p>
<button onclick="runAction('/custom_actions/warm_up', 'out_warm_up')">Start warm_up</button>
<pre id="out_warm_up" style="margin-top:1em"></pre>

<script>
function runAction(url, outId) {
    document.getElementById(outId).textContent = "Computing...";
    fetch(url, {method: "POST"})
        .then(r => r.json())
        .then(d => document.getElementById(outId).textContent = d.command || JSON.stringify(d, null, 2))
        .catch(e => document.getElementById(outId).textContent = "Error: " + e);
}
</script>
</body>
</html>'''


def _register_custom_actions(app, environment):
    # Shared single lock — only one delete action (of either kind) may run at a time
    action_lock = threading.Lock()
    action_lock_holder = [None]  # mutable cell so inner functions can read/write it

    def _make_action_logger(name, filename):
        logger = logging.getLogger(name)
        logger.setLevel(logging.INFO)
        logger.propagate = False
        handler = logging.FileHandler(filename)
        handler.setFormatter(logging.Formatter('%(asctime)s %(levelname)s %(message)s'))
        logger.addHandler(handler)
        return logger

    phases_log = _make_action_logger('phases', 'locust_results_phases.log')
    normal_run_ended = [False]
    post_run_started = [False]
    auto_execute_timer = [None]
    phases_log.info('START Normal run')

    def _end_normal_run_if_needed():
        if not normal_run_ended[0]:
            phases_log.info('END Normal run')
            normal_run_ended[0] = True

    def _start_post_run():
        phases_log.info('START PostRun')
        post_run_started[0] = True

    @app.route('/custom_actions', methods=['GET'])
    def custom_actions():
        return render_template_string(_CUSTOM_ACTIONS_HTML)

    # ---------------------------------------------------------------------------
    # Shared range computation for the 10%-deletion actions
    # ---------------------------------------------------------------------------

    def _compute_delete_range():
        # Random starting offset: between 10% and 30% from the beginning of the
        # index, so the deletion range doesn't always start at the minimum key.
        skip_docs = round(COLLECTION_COUNT * uniform(0.10, 0.30))

        # Number of documents to delete: 10% of the configured collection size.
        # Expressed via AVG_DOC_BYTES to make the relationship to data volume explicit,
        # even though the avg size cancels out algebraically.
        docs_to_delete = round(COLLECTION_COUNT * AVG_DOC_BYTES * 0.10 / AVG_DOC_BYTES)

        # Covered index scans: projection excludes _id so only the shardKey index is
        # read, with no access to the underlying documents.
        index_only = {'shardKey': 1, '_id': 0}
        start_key = next(
            collection.with_options(read_preference=ReadPreference.SECONDARY_PREFERRED).find(
                {},
                index_only,
                sort=[('shardKey', 1)],
            ).skip(skip_docs).limit(1))['shardKey']

        cutoff_key = next(
            collection.with_options(read_preference=ReadPreference.SECONDARY_PREFERRED).find(
                {},
                index_only,
                sort=[('shardKey', 1)],
            ).skip(skip_docs + docs_to_delete).limit(1))['shardKey']

        return docs_to_delete, start_key, cutoff_key

    # ---------------------------------------------------------------------------
    # deleteMany - 10% of the data
    # ---------------------------------------------------------------------------

    def _run_delete_many(coll_name, start_key, cutoff_key, command, on_complete=None):
        _end_normal_run_if_needed()
        phases_log.info('START delete_many')
        phases_log.info(f'command={command}')
        try:
            result = collection.database.command(
                'delete', coll_name, deletes=[{
                    'q': {
                        'shardKey': {
                            '$gte': start_key,
                            '$lt': cutoff_key
                        }
                    },
                    'limit': 0
                }])
            phases_log.info(f'RESULT {result}')
        except Exception as exc:
            phases_log.error(f'FAILED {exc}')
        finally:
            phases_log.info('END delete_many')
            action_lock_holder[0] = None
            action_lock.release()
            if on_complete:
                on_complete()

    def _deleteMany_10_pct(execute, on_complete=None):
        try:
            planned, start_key, cutoff_key = _compute_delete_range()
            db_name = collection.database.name
            coll_name = collection.name
            command = (
                f'db.getSiblingDB("{db_name}").runCommand({{'
                f' delete: "{coll_name}",'
                f' deletes: [{{ q: {{ shardKey: {{ $gte: {json.dumps(start_key)}, $lt: {json.dumps(cutoff_key)} }} }}, limit: 0 }}]'
                f' }})')
            if execute:
                if not action_lock.acquire(blocking=False):
                    return {'error': f'{action_lock_holder[0]} is already running'}, 409
                action_lock_holder[0] = 'deleteMany_10_pct'
                threading.Thread(
                    target=_run_delete_many,
                    args=(coll_name, start_key, cutoff_key, command),
                    kwargs={
                        'on_complete': on_complete
                    },
                    daemon=True,
                ).start()
                return {'status': 'started', 'planned': planned, 'command': command}, 200
            return {'planned': planned, 'command': command}, 200
        except Exception as e:
            return {'error': str(e)}, 500

    @app.route('/custom_actions/deleteMany_10_pct', methods=['POST'])
    def deleteMany_10_pct():
        data, status = _deleteMany_10_pct(
            flask_request.args.get('execute', '').lower() in ('true', '1', 'yes'))
        return jsonify(data), status

    # ---------------------------------------------------------------------------
    # fastBulkDelete - 10% of the data
    # ---------------------------------------------------------------------------

    def _run_fast_bulk_delete(coll_name, start_key, cutoff_key, command, on_complete=None):
        _end_normal_run_if_needed()
        phases_log.info('START fastBulkDelete')
        phases_log.info(f'command={command}')
        try:
            result = collection.database.command('fastBulkDelete', coll_name,
                                                 filterIndexName='shardKey',
                                                 lowerBound={'shardKey': start_key},
                                                 upperBound={'shardKey': cutoff_key})
            phases_log.info(f'RESULT {result}')
        except Exception as exc:
            phases_log.error(f'FAILED {exc}')
        finally:
            phases_log.info('END fastBulkDelete')
            action_lock_holder[0] = None
            action_lock.release()
            if on_complete:
                on_complete()

    def _fastBulkDelete_10_pct(execute, on_complete=None):
        try:
            planned, start_key, cutoff_key = _compute_delete_range()
            db_name = collection.database.name
            coll_name = collection.name
            command = (f'db.getSiblingDB("{db_name}").runCommand({{'
                       f' fastBulkDelete: "{coll_name}",'
                       f' filterIndexName: "shardKey",'
                       f' lowerBound: {{ shardKey: {json.dumps(start_key)} }},'
                       f' upperBound: {{ shardKey: {json.dumps(cutoff_key)} }}'
                       f' }})')
            if execute:
                if not action_lock.acquire(blocking=False):
                    return {'error': f'{action_lock_holder[0]} is already running'}, 409
                action_lock_holder[0] = 'fastBulkDelete_10_pct'
                threading.Thread(
                    target=_run_fast_bulk_delete,
                    args=(coll_name, start_key, cutoff_key, command),
                    kwargs={
                        'on_complete': on_complete
                    },
                    daemon=True,
                ).start()
                return {'status': 'started', 'planned': planned, 'command': command}, 200
            return {'planned': planned, 'command': command}, 200
        except Exception as e:
            return {'error': str(e)}, 500

    @app.route('/custom_actions/fastBulkDelete_10_pct', methods=['POST'])
    def fastBulkDelete_10_pct():
        data, status = _fastBulkDelete_10_pct(
            flask_request.args.get('execute', '').lower() in ('true', '1', 'yes'))
        return jsonify(data), status

    # ---------------------------------------------------------------------------
    # warm_up - covered index scan over every secondary index
    # ---------------------------------------------------------------------------

    def _do_warm_up_work():
        _end_normal_run_if_needed()
        phases_log.info('START WarmUp')
        try:
            coll = collection.with_options(read_preference=ReadPreference.SECONDARY_PREFERRED)
            for field in SECONDARY_INDEX_FIELDS:
                start = perf_counter_ns()
                # Hint forces a full scan of this index; count_documents executes a
                # server-side $count over the matched index entries (covered).
                count = coll.count_documents({}, hint=[(field, 1)])
                elapsed_s = (perf_counter_ns() - start) / 1e9
                phases_log.info(f'index={field} count={count} elapsed={elapsed_s:.1f}s')
        except Exception as exc:
            phases_log.error(f'FAILED {exc}')
        finally:
            phases_log.info('END WarmUp')

    def _run_warm_up(on_complete=None):
        try:
            _do_warm_up_work()
        finally:
            action_lock_holder[0] = None
            action_lock.release()
            if on_complete:
                on_complete()

    def _warm_up(on_complete=None):
        if not action_lock.acquire(blocking=False):
            return {'error': f'{action_lock_holder[0]} is already running'}, 409
        action_lock_holder[0] = 'warm_up'
        threading.Thread(
            target=_run_warm_up,
            kwargs={
                'on_complete': on_complete
            },
            daemon=True,
        ).start()
        return {'status': 'started', 'fields': SECONDARY_INDEX_FIELDS}, 200

    @app.route('/custom_actions/warm_up', methods=['POST'])
    def warm_up():
        data, status = _warm_up()
        return jsonify(data), status

    # ---------------------------------------------------------------------------
    # Auto-execute: fire one action after a fixed delay if --auto-execute is set
    # ---------------------------------------------------------------------------

    def _auto_execute_action():
        action_name = environment.parsed_options.auto_execute
        logging.info(f'auto-execute: launching {action_name}')

        def _on_complete():
            _start_post_run()
            logging.info(f'auto-execute: quitting in {_AUTO_EXECUTE_QUIT_DELAY_SECS}s')
            threading.Timer(_AUTO_EXECUTE_QUIT_DELAY_SECS, environment.runner.quit).start()

        if action_name == 'deleteMany_10_pct':
            data, status = _deleteMany_10_pct(execute=True, on_complete=_on_complete)
        elif action_name == 'fastBulkDelete_10_pct':
            data, status = _fastBulkDelete_10_pct(execute=True, on_complete=_on_complete)

        if status != 200:
            logging.error(f'auto-execute {action_name}: failed to start: {data.get("error")}')

    def _start_auto_execute_timer():
        delay = environment.parsed_options.auto_execute_delay
        logging.info(
            f'Scheduling auto-execute of {environment.parsed_options.auto_execute} in {delay}s')
        t = threading.Timer(delay, _auto_execute_action)
        t.daemon = True
        t.start()
        auto_execute_timer[0] = t

    @events.spawning_complete.add_listener
    def on_spawning_complete(**_):
        if environment.parsed_options.warm_up:

            def _after_warm_up():
                if environment.parsed_options.auto_execute:
                    _start_auto_execute_timer()
                else:
                    _start_post_run()

            data, status = _warm_up(on_complete=_after_warm_up)
            if status != 200:
                logging.error(f'warm-up failed to start: {data.get("error")}')
        elif environment.parsed_options.auto_execute:
            _start_auto_execute_timer()

    @events.quitting.add_listener
    def on_quitting(**_):
        _end_normal_run_if_needed()
        if post_run_started[0]:
            phases_log.info('END PostRun')
        if auto_execute_timer[0] is not None:
            auto_execute_timer[0].cancel()