testsuite/python/tests/test_100_3.py

############################################################################
# Copyright (C) SchedMD LLC.
############################################################################
import logging
import math
import re
import threading
import time
from datetime import datetime

import pytest

import atf

pytestmark = pytest.mark.slow

# Global variables
cluster = "test_cluster1"

# total number of rows created for all tables in purge type
row_count_total = 500000

# purge for those rows should happen in less than this time
max_perf_time_default = 60

# From 26.11 every job refers to a KV document in its own kv_table row, the
# most a job purge can have to purge from the kv table
kv_supported = atf.get_version("sbin/slurmdbd") >= (26, 11)
job_kv_col = ",kv_hash_inx" if kv_supported else ""
job_kv_val = ",@seq" if kv_supported else ""

table_gen_info = [
    {
        "type": "Event",
        "tables": [
            {
                "name": f"{cluster}_event_table",
                "sql": "(time_start,time_end,reason) values (@seq,@seq,'')",
                # Ex. override row count and set fixed amount if needed
                #  'row_count'     : 100,
            },
        ],
        "max_perf_time": max_perf_time_default,
    },
    {
        "type": "Job",
        "tables": [
            {
                "name": f"{cluster}_job_table",
                "sql": f"(cpus_req,job_name,id_assoc,id_job,id_resv,id_wckey,id_user,id_group,het_job_id,het_job_offset,state_reason_prev,nodes_alloc,\\`partition\\`,priority,state,time_end,env_hash_inx,script_hash_inx{job_kv_col}) values (0,'',0,@seq,0,0,0,0,0,0,0,0,'',0,0,@seq,@seq,@seq{job_kv_val})",
            },
            # job_table will be used to seed remaining job tables to speed things up
            {
                "name": f"{cluster}_job_env_table",
                "sql": f"(hash_inx,env_hash) select env_hash_inx,env_hash_inx from {cluster}_job_table",
                "from_template": True,
            },
            {
                "name": f"{cluster}_job_script_table",
                "sql": f"(hash_inx,script_hash) select script_hash_inx,script_hash_inx from {cluster}_job_table",
                "from_template": True,
            },
            {
                "name": f"{cluster}_kv_table",
                "sql": f"(hash_inx,kv_hash) select kv_hash_inx,kv_hash_inx from {cluster}_job_table",
                "from_template": True,
                "minversion": (26, 11),
            },
        ],
        "max_perf_time": max_perf_time_default,
    },
    {
        "type": "Resv",
        "tables": [
            {
                "name": f"{cluster}_resv_table",
                "sql": "(id_resv,time_end,resv_name) values (@seq,@seq,'')",
            },
        ],
        "max_perf_time": max_perf_time_default,
    },
    {
        "type": "Step",
        "tables": [
            {
                "name": f"{cluster}_step_table",
                "sql": "(job_db_inx,id_step,nodelist,nodes_alloc,state,step_name,task_cnt,time_end) values (@seq,@seq,'',0,0,'',0,@seq)",
            },
        ],
        "max_perf_time": max_perf_time_default,
    },
    {
        "type": "Suspend",
        "tables": [
            {
                "name": f"{cluster}_suspend_table",
                "sql": "(job_db_inx,time_end,id_assoc) values (@seq,@seq,0)",
            },
        ],
        "max_perf_time": max_perf_time_default,
    },
    {
        "type": "TXN",
        "tables": [
            {
                "name": "txn_table",
                "sql": f"(timestamp,cluster,action,name,actor) values (@seq,'{cluster}',0,'','')",
            },
        ],
        "max_perf_time": max_perf_time_default,
    },
    {
        "type": "Usage",
        "tables": [
            # qos usage tables added in 24.11 but not purged until 25.05.3, list them first for row count calculation later
            {
                "name": f"{cluster}_qos_usage_hour_table",
                "sql": "(creation_time,mod_time,id,time_start) values (@seq,@seq,1,@seq)",
                "minversion": (25, 5, 3),
            },
            # qos_usage_hour_table will be used to seed remaining qos usage tables to speed things up
            {
                "name": f"{cluster}_qos_usage_day_table",
                "sql": f"(creation_time,mod_time,id,time_start) select creation_time,mod_time,id,time_start from {cluster}_qos_usage_hour_table",
                "from_template": True,
                "minversion": (25, 5, 3),
            },
            {
                "name": f"{cluster}_qos_usage_month_table",
                "sql": f"(creation_time,mod_time,id,time_start) select creation_time,mod_time,id,time_start from {cluster}_qos_usage_hour_table",
                "from_template": True,
                "minversion": (25, 5, 3),
            },
            {
                "name": f"{cluster}_assoc_usage_hour_table",
                "sql": "(creation_time,mod_time,id,time_start) values (@seq,@seq,@seq,@seq)",
            },
            # assoc_usage_hour_table will be used to seed remaining assoc usage tables to speed things up
            {
                "name": f"{cluster}_assoc_usage_day_table",
                "sql": f"(creation_time,mod_time,id,time_start) select creation_time,mod_time,id,time_start from {cluster}_assoc_usage_hour_table",
                "from_template": True,
            },
            {
                "name": f"{cluster}_assoc_usage_month_table",
                "sql": f"(creation_time,mod_time,id,time_start) select creation_time,mod_time,id,time_start from {cluster}_assoc_usage_hour_table",
                "from_template": True,
            },
            {
                "name": f"{cluster}_wckey_usage_hour_table",
                "sql": f"(creation_time,mod_time,id,time_start) select creation_time,mod_time,id,time_start from {cluster}_assoc_usage_hour_table",
                "from_template": True,
            },
            {
                "name": f"{cluster}_wckey_usage_day_table",
                "sql": f"(creation_time,mod_time,id,time_start) select creation_time,mod_time,id,time_start from {cluster}_assoc_usage_hour_table",
                "from_template": True,
            },
            {
                "name": f"{cluster}_wckey_usage_month_table",
                "sql": f"(creation_time,mod_time,id,time_start) select creation_time,mod_time,id,time_start from {cluster}_assoc_usage_hour_table",
                "from_template": True,
            },
            {
                "name": f"{cluster}_usage_hour_table",
                "sql": f"(creation_time,mod_time,id_tres,time_start) select creation_time,mod_time,id,time_start from {cluster}_assoc_usage_hour_table",
                "from_template": True,
            },
            {
                "name": f"{cluster}_usage_day_table",
                "sql": f"(creation_time,mod_time,id_tres,time_start) select creation_time,mod_time,id,time_start from {cluster}_assoc_usage_hour_table",
                "from_template": True,
            },
            {
                "name": f"{cluster}_usage_month_table",
                "sql": f"(creation_time,mod_time,id_tres,time_start) select creation_time,mod_time,id,time_start from {cluster}_assoc_usage_hour_table",
                "from_template": True,
            },
        ],
        "max_perf_time": max_perf_time_default,
    },
]


@pytest.fixture(scope="module", autouse=True)
def setup():
    """Test setup with required configurations."""

    atf.require_accounting(modify=True)
    atf.require_config_parameter_includes("AccountingStorageEnforce", "associations")
    atf.require_config_parameter("AllowNoDefAcct", None, source="slurmdbd")

    # if any purge types are to be skipped, comment them out below
    atf.require_config_parameter("PurgeEventAfter", "1h", source="slurmdbd")
    atf.require_config_parameter("PurgeJobAfter", "1h", source="slurmdbd")
    atf.require_config_parameter("PurgeResvAfter", "1h", source="slurmdbd")
    atf.require_config_parameter("PurgeStepAfter", "1h", source="slurmdbd")
    atf.require_config_parameter("PurgeSuspendAfter", "1h", source="slurmdbd")
    atf.require_config_parameter("PurgeTXNAfter", "1h", source="slurmdbd")
    atf.require_config_parameter("PurgeUsageAfter", "1h", source="slurmdbd")

    # debug2 really only required
    atf.require_config_parameter("DebugLevel", "debug2", source="slurmdbd")
    # these can be helpful when debugging/tracing
    # atf.require_config_parameter("DebugLevel", "debug4", source="slurmdbd")
    # atf.require_config_parameter("DebugFlags", f"{atf.get_config(live=False, source="slurmdbd", quiet=True)["DebugFlags"]},DB_USAGE,DB_ARCHIVE", source="slurmdbd")

    # so database will exist
    atf.require_slurm_running()


@pytest.fixture(scope="function", autouse=True)
def setup_db(sql_statement_repeat):

    atf.run_command(
        f"sacctmgr -i add cluster {cluster}",
        user=atf.properties["slurm-user"],
        fatal=True,
        # quiet=True,
    )

    yield

    atf.run_command(
        f"sacctmgr -i remove cluster {cluster}",
        user=atf.properties["slurm-user"],
        # quiet=True,
    )


# we need to pass in a mutable object (list with index) so the thread results can be returned
def run_command_bg(command, result, index, **run_command_kwargs):
    result[index] = atf.run_command(command, **run_command_kwargs)


def test_purge_slurm_db_tables(sql_statement_repeat):
    """Test purging large number of rows"""

    # rollup beginning/ending log signatures
    rollup_beg_pat = "running rollup"
    rollup_end_pat = "everything rolled"

    dbd_log = atf.get_config_parameter(
        "LogFile", live=False, quiet=True, source="slurmdbd"
    )

    for entry in table_gen_info:
        if (
            atf.get_config_parameter(
                f"Purge{entry['type']}After", live=False, quiet=True, source="slurmdbd"
            )
            is None
        ):
            continue

        # stop so db can be altered
        atf.stop_slurmdbd()

        # make entries 2hrs old
        two_hours_ago = int(time.time()) - 2 * 3600

        count_sql = ""
        table_count = len(entry["tables"])
        for table_info in entry["tables"]:
            if (
                "minversion" in table_info
                and atf.get_version("sbin/slurmdbd") < table_info["minversion"]
            ):
                table_count -= 1
                continue

            # if row_count was given then use it, otherwise make each table equal size
            if "row_count" in table_info:
                row_count = int(table_info["row_count"])
            else:
                row_count = int(math.ceil(row_count_total / table_count))

            # populate the table
            if "from_template" in table_info and table_info["from_template"]:
                # data copied from another template table so no need to repeat insert
                # this makes things faster
                seq_start = 1
                seq_end = 1
            else:
                seq_start = two_hours_ago - row_count + 1
                seq_end = two_hours_ago
            mysql_command = sql_statement_repeat
            mysql_command += f" -e \"call statement_repeat(\\\"insert into {table_info['name']} {table_info['sql']}\\\", {seq_start}, {seq_end}, 1, 1)\""
            logging.info(
                f"Populating database with {row_count} rows in {table_info['name']}"
            )
            atf.run_command(
                mysql_command,
                user=atf.properties["slurm-user"],
                fatal=True,
            )

            count_sql += (
                f"{'+' if count_sql else ''}(select count(*) from {table_info['name']})"
            )

        # save starting sum of table row counts
        mysql_command = sql_statement_repeat + f' -Ns -e "select {count_sql}"'
        starting_row_count = atf.run_command_output(
            mysql_command,
            user=atf.properties["slurm-user"],
            # quiet=True,
            fatal=True,
        )

        # fabricate "last ran" time of 2hrs ago so hourly rollup/purge will be triggered
        mysql_command = (
            sql_statement_repeat
            + f' -e "delete ignore from {cluster}_last_ran_table; insert into {cluster}_last_ran_table (hourly_rollup, daily_rollup, monthly_rollup) values ({two_hours_ago}, {two_hours_ago}, {two_hours_ago})"'
        )
        atf.run_command(
            mysql_command,
            user=atf.properties["slurm-user"],
            fatal=True,
        )

        logging.info(
            f"Waiting {entry['max_perf_time']}s for {entry['type']} purge to complete (happens as part of rollup)"
        )

        # start watching the log before starting slurmdbd so all desired output can be captured
        results = [None]
        index = 0
        tail_thread = threading.Thread(
            target=run_command_bg,
            args=(
                # use "-n 1" so we don't match rollup_end_pat from previous run
                f"tail -n 1 -F {dbd_log} | grep -E --line-buffered -i -m2 '{rollup_beg_pat}|{rollup_end_pat}'",
                results,
                index,
            ),
            kwargs={
                "timeout": entry["max_perf_time"],
                "quiet": True,
                "user": atf.properties["slurm-user"],
            },
        )
        tail_thread.start()

        # wait for thread to start
        while tail_thread.ident is None:
            time.sleep(0.1)

        atf.start_slurmdbd()

        # wait for rollup (purge) to finish or timeout
        tail_thread.join()

        logging.info(
            f"Rollup lines identified in {dbd_log}:\n{results[index]['stdout'].rstrip()}"
        )

        search_result = re.search(
            rf"(\d\d\d\d-\d\d-\d\dT\d\d:\d\d:\d\d\.\d\d\d).*{rollup_beg_pat}.*\s*.*(\d\d\d\d-\d\d-\d\dT\d\d:\d\d:\d\d\.\d\d\d).*{rollup_end_pat}",
            results[index]["stdout"],
            re.IGNORECASE,
        )

        assert (
            search_result
        ), f"slurmdbd should be able to purge {entry['type']} tables in less than {entry['max_perf_time']}s"

        # count the actual number of rows removed by the purge
        mysql_command = sql_statement_repeat + f' -Ns -e "select {count_sql}"'
        ending_row_count = atf.run_command_output(
            mysql_command,
            user=atf.properties["slurm-user"],
            # quiet=True,
            fatal=True,
        )
        rows_removed = int(starting_row_count) - int(ending_row_count)

        start_obj = datetime.strptime(search_result.group(1), "%Y-%m-%dT%H:%M:%S.%f")
        end_obj = datetime.strptime(search_result.group(2), "%Y-%m-%dT%H:%M:%S.%f")
        time_sec = (end_obj - start_obj).total_seconds()
        rate = int(rows_removed / time_sec)

        # debug: list tables with leftover rows
        mysql_command = (
            sql_statement_repeat
            + f" -e \"select rpad(table_name,40,' '),lpad(table_rows,6,' ') from information_schema.tables where table_schema=database() and table_name like '%_{entry['type'].lower()}_%' and table_rows>0\""
        )
        atf.run_command_output(mysql_command, user=atf.properties["slurm-user"])

        logging.info(
            f"{entry['type']} purge took {time_sec}s to remove {rows_removed} rows from {table_count} tables ({rate} rows/sec)"
        )


@pytest.mark.skipif(
    atf.get_version("sbin/slurmdbd") < (26, 11)
    or atf.get_version("bin/sacctmgr") < (26, 11),
    reason="Ticket 24919: DisableArchiveAnalyze was added in 26.11, and show "
    "config is rendered by sacctmgr, so an older client omits the keypair",
)
def test_disable_archive_analyze_config():
    """DisableArchiveAnalyze must default to off and be honored once set.

    The option gates the ANALYZE TABLE slurmdbd runs after each archive/purge
    cycle.  ANALYZE is only logged when it fails, so there is no success
    signature to match on; what is checked here is that the keypair reaches
    sacctmgr show config with the documented default, that setting it is
    reported back.

    It deliberately does not trigger a rollup to exercise the purge path.
    sacctmgr rollup takes no cluster filter, and this module configures every
    Purge*After, so a rollup here runs an archive/purge cycle over every
    cluster in the shared database -- including the multi-day-old rows other
    test files are concurrently relying on.
    """
    for configured, expected in ((None, "no"), ("yes", "yes")):
        atf.require_config_parameter(
            "DisableArchiveAnalyze", configured, source="slurmdbd"
        )
        reported = atf.get_config_parameter(
            "DisableArchiveAnalyze", live=True, source="sacctmgr"
        )
        assert reported is not None and reported.lower() == expected, (
            f"sacctmgr show config reports DisableArchiveAnalyze={reported}, "
            f"expected {expected}"
        )