datastore/ci/jobs/stress_job.py

import csv
import logging
import os
import sys
from pathlib import Path
from typing import List, Tuple

from ci.jobs.scripts.clickhouse_service import ClickHouseService
from ci.jobs.scripts.docker_image import DockerImage
from ci.jobs.scripts.log_parser import FuzzerLogParser
from ci.praktika.info import Info
from ci.praktika.result import Result
from ci.praktika.utils import Shell, Utils


def sanitize_test_result_line(line: str) -> str:
    # Drop bare CR in addition to escaping NUL. The writer escapes
    # `\0\t\n` into backslash forms (`escape_tsv_info`) but not `\r`,
    # and the apt-get / dpkg progress frames captured into the
    # `Hung check failed` info field by `_ensure_lldb_installed` use
    # bare CR to overwrite the previous "(Reading database ... N%)"
    # frame. Left in place, those CRs are translated to LF by
    # universal-newlines mode and fragment the row.
    return line.replace("\0", "\\0").replace("\r", "")


def read_test_results(results_path: Path, with_raw_logs: bool = True):
    """Parse the stress-job `test_results.tsv` file.

    Returns `(results, malformed)`:
      - `results`: valid `Result` rows.
      - `malformed`: list of `(line_number, raw_first_cell)` tuples for
        rows that had fewer than 2 tab-separated cells.

    Malformed rows are skipped rather than fatal. They commonly appear
    when something writes stray output into the file (e.g. an
    `apt-get install` log line leaking into the result directory), and
    raising on the first one would discard every valid row above and
    below it — including the real failure that triggered the job.
    Callers should surface the count back to investigators via a
    separate `Result` entry so the corruption is still noticed.
    """
    results = []
    malformed = []
    # Split on LF only (not on CR or CRLF) so bare CR bytes that the
    # writer failed to escape stay inside their row instead of
    # fragmenting it. `sanitize_test_result_line` then strips those
    # CRs before csv parsing. Python's default text-mode `open` would
    # turn every bare CR into LF (universal-newlines), and `newline=""`
    # still splits lines on CR — both behaviours produce the "row
    # exploded into many fragments" failure observed on PR #105243
    # Stress test (arm_debug).
    with open(results_path, "rb") as descriptor:
        raw = descriptor.read().decode("utf-8", errors="replace")
    lines = raw.split("\n")
    reader = csv.reader(
        (sanitize_test_result_line(line) for line in lines),
        delimiter="\t",
    )
    for line_number, line in enumerate(reader, start=1):
        # Blank lines (typically a trailing newline at end of file, or
        # a separator artifact between writes) carry no information —
        # skip them silently.
        if not line:
            continue
        if len(line) < 2:
            malformed.append((line_number, line[0]))
            continue
        name = line[0]
        status = line[1]
        time = None
        if len(line) >= 3 and line[2] and line[2] != "\\N":
            # The value can be empty, but when it's not,
            # it's the time spent on the test
            try:
                time = float(line[2])
            except ValueError:
                pass

        result = Result(name, status, duration=time)
        if not (result.is_ok() or result.is_failure() or result.is_error()):
            # Unknown status — treat as a malformed row rather than
            # aborting the whole file. Aborting would re-introduce the
            # `Unknown job error` failure mode this parser is here to
            # eliminate: a single unexpected-status row would discard
            # every valid neighbour, including the real failure that
            # triggered the job. (The pre-existing assert referenced
            # `is_failure`/`is_error` as attributes — unbound method
            # objects, always truthy — so it never fired and let any
            # status through silently. Fix to call the methods.)
            malformed.append((line_number, line[0]))
            continue
        if len(line) == 4 and line[3]:
            # The value can be empty, but when it's not,
            # the 4th value is a pythonic list, e.g. ['file1', 'file2']
            if with_raw_logs:
                # Python does not support TSV, so we unescape manually.
                # The writer (`escape_tsv_info`) emits `\\r` for CR, so
                # unescape it here too. Without this, dpkg/apt-get
                # progress markers in the info field would leak the
                # literal two-character `\r` sequence into the displayed
                # log.
                result.set_info(
                    line[3]
                    .replace("\\t", "\t")
                    .replace("\\r", "\r")
                    .replace("\\n", "\n")
                )
            else:
                result.set_info(line[3])
        results.append(result)
    return results, malformed


def _format_malformed_summary(
    results_path: Path,
    malformed: List[Tuple[int, str]],
) -> str:
    sample = malformed[0]
    preview = sample[1][:200]
    return (
        f"{len(malformed)} malformed row(s) in {results_path} skipped; "
        f"first at line {sample[0]}: {preview!r}"
    )


def get_additional_envs(info, check_name: str) -> List[str]:
    from ci.jobs.ci_utils import is_extended_run

    result = []
    # some cloud-specific features require feature flags enabled
    # so we need this ENV to be able to disable the randomization
    # of feature flags
    result.append("RANDOMIZE_KEEPER_FEATURE_FLAGS=1")
    if "azure" in check_name:
        result.append("USE_AZURE_STORAGE_FOR_MERGE_TREE=1")

    if "s3" in check_name:
        result.append("USE_S3_STORAGE_FOR_MERGE_TREE=1")

    result.append(
        f"STRESS_GLOBAL_TIME_LIMIT={'3600' if is_extended_run() else '1200'}"
    )

    return result


def get_run_command(
    build_path: Path,
    result_path: Path,
    repo_tests_path: Path,
    server_log_path: Path,
    cores_path: Path,
    additional_envs: List[str],
    image: DockerImage,
    upgrade_check: bool,
) -> str:
    envs = [f"-e {e}" for e in additional_envs]
    env_str = " ".join(envs)

    if upgrade_check:
        run_script = "/repo/tests/docker_scripts/upgrade_runner.sh"
    else:
        run_script = "/repo/tests/docker_scripts/stress_runner.sh"

    cmd = (
        "docker run --cap-add=SYS_PTRACE "
        # For dmesg and sysctl
        "--privileged "
        # azurite-rs (in-process Azure Blob Storage emulator) needs many fds under parallel load
        "--ulimit nofile=1048576:1048576 "
        # a static link, don't use S3_URL or S3_DOWNLOAD
        "-e S3_URL='https://s3.amazonaws.com/clickhouse-datasets' "
        "--tmpfs /tmp/clickhouse:mode=1777 "
        f"--volume={build_path}:/package_folder "
        f"--volume={result_path}:/test_output "
        f"--volume={repo_tests_path}/..:/repo "
        f"--volume={server_log_path}:/var/log/clickhouse-server "
        f"--volume={cores_path}:/cores "
        f"{env_str} {image} {run_script}"
    )

    return cmd


def process_results(
    result_directory: Path, server_log_path: Path
) -> Tuple[str, str, List[Result], List[Path]]:
    test_results = []
    additional_files = []
    # Just upload all files from result_folder.
    # If task provides processed results, then it's responsible for content
    # of result_folder.
    if result_directory.exists():
        additional_files = [p for p in result_directory.iterdir() if p.is_file()]

    if server_log_path.exists():
        additional_files = additional_files + [
            p for p in server_log_path.iterdir() if p.is_file()
        ]

    results_path = result_directory / "test_results.tsv"
    malformed: List[Tuple[int, str]] = []
    try:
        test_results, malformed = read_test_results(results_path, True)
        if len(test_results) == 0 and not malformed:
            raise ValueError("Empty results")
    except Exception as e:
        test_results = [
            Result(
                name="Unknown job error",
                status=Result.Status.ERROR,
                info=f"Cannot parse test_results.tsv ({e})",
            )
        ]
        return test_results, additional_files

    if not test_results:
        # The file existed and had content, but every row was malformed.
        # Surface the corruption directly instead of "Unknown job error".
        test_results = [
            Result(
                name="Corrupt test_results.tsv",
                status=Result.Status.ERROR,
                info=_format_malformed_summary(results_path, malformed),
            )
        ]
        return test_results, additional_files

    if malformed:
        # Some rows were unreadable but valid rows survived. Add a
        # FAIL entry so investigators see the corruption without
        # losing the real test results (in particular, the
        # `Hung check failed, possible deadlock found` row that
        # otherwise gets swallowed when the file is polluted by
        # stray output such as `apt-get install` logs).
        test_results.append(
            Result(
                name="Corrupt test_results.tsv",
                status=Result.Status.FAIL,
                info=_format_malformed_summary(results_path, malformed),
            )
        )

    return test_results, additional_files


def run_stress_test(upgrade_check: bool = False) -> None:
    info = Info()
    logging.basicConfig(level=logging.INFO)

    stopwatch = Utils.Stopwatch()
    temp_path = Path(Utils.cwd()) / "ci/tmp"
    repo_tests_path = Path(Utils.cwd()) / "tests"

    check_name = sys.argv[1] if len(sys.argv) > 1 else os.getenv("CHECK_NAME")
    assert (
        check_name
    ), "Check name must be provided as an input arg or in CHECK_NAME env"

    packages_path = temp_path

    docker_image = DockerImage.get_docker_image("clickhouse/stress-test").pull_image()

    server_log_path = temp_path / "server_log"
    server_log_path.mkdir(parents=True, exist_ok=True)

    result_path = temp_path / "result_path"
    result_path.mkdir(parents=True, exist_ok=True)

    cores_path = temp_path / "cores"
    cores_path.mkdir(parents=True, exist_ok=True)

    additional_envs = get_additional_envs(info, check_name)

    run_command = get_run_command(
        packages_path,
        result_path,
        repo_tests_path,
        server_log_path,
        cores_path,
        additional_envs,
        docker_image,
        upgrade_check,
    )
    logging.info("Going to run stress test: %s", run_command)

    exit_code = Shell.run(run_command)

    Utils.fix_ownership_after_docker(temp_path, docker_image)

    core_files = ClickHouseService.collect_cores(cores_path)

    is_oom = False

    if Path(result_path / "dmesg.log").is_file():
        is_oom = Shell.check(
            "grep -q -F -e 'Out of memory: Killed process' -e 'oom_reaper: reaped process' -e 'oom-kill:constraint=CONSTRAINT_NONE' "
            f"{result_path}/dmesg.log"
        )

    # Check for OOM (signal 9) in server logs
    if server_log_path.exists():
        server_log_oom = Shell.check(
            f"rg -Fqa ' <Fatal> Application: Child process was terminated by signal 9' "
            f"{server_log_path}/clickhouse-server*.log"
        )
        is_oom = is_oom or server_log_oom

    # Generate fatal.log from all server logs
    fatal_log = result_path / "fatal.log"
    if server_log_path.exists():
        Shell.check(
            f"rg --text '\\s<Fatal>\\s' {server_log_path}/clickhouse-server*.log > {fatal_log}"
        )

    test_results, additional_logs = process_results(result_path, server_log_path)

    server_died = False
    failed_results = []
    for test_result in test_results:
        if test_result.name == "Server died":
            server_died = True
            continue
        if not test_result.is_ok():
            failed_results.append(test_result)

    if server_died:
        # Build log pairs for each replica: (replica_name, server_err_log, stderr_log)
        # Main replica: all *.err.* logs without sc1/sc2 in name, paired with stderr.log
        # sc1/sc2: their dedicated server log + matching stderr log
        replica_log_pairs = []

        main_stderr = result_path / "stderr.log"
        if server_log_path.exists():
            main_server_logs = sorted(
                p
                for p in server_log_path.iterdir()
                if p.is_file()
                and ".err." in p.name
                and "sc1" not in p.name
                and "sc2" not in p.name
            )
            for log_file in main_server_logs:
                replica_log_pairs.append(("main", log_file, main_stderr))

        for sc in ("sc1", "sc2"):
            sc_server_log = server_log_path / f"clickhouse-server-{sc}.err.log"
            sc_stderr = result_path / f"stderr-{sc}.log"
            if sc_server_log.exists():
                replica_log_pairs.append((sc, sc_server_log, sc_stderr))

        if not replica_log_pairs:
            failed_results.append(
                Result.create_from(
                    name="Unknown error",
                    info="no server logs found",
                    status=Result.Status.FAIL,
                )
            )
        else:
            definitive_result = None
            fallback_result = None

            for replica_name, server_log_file, stderr_log in replica_log_pairs:
                log_parser = FuzzerLogParser(
                    server_log=server_log_file,
                    stderr_log=str(stderr_log) if stderr_log.exists() else "",
                    fuzzer_log="",
                )
                try:
                    name, description, files = log_parser.parse_failure()
                    file_pair_info = f"Log files: {server_log_file.name}"
                    if stderr_log.exists():
                        file_pair_info += f", {stderr_log.name}"
                    description = f"{file_pair_info}\n{description}"
                    if name != FuzzerLogParser.UNKNOWN_ERROR:
                        definitive_result = (name, description, files)
                        break
                    if fallback_result is None:
                        fallback_result = (name, description, files)
                except Exception as e:
                    print(
                        f"ERROR: Failed to parse failure logs for {replica_name} "
                        f"({server_log_file.name}): {e}\n"
                        f"Server logs should still be collected."
                    )

            result = definitive_result or fallback_result
            if result:
                name, description, files = result
                failed_results.append(
                    Result.create_from(
                        name=name,
                        info=description,
                        status=Result.Status.FAIL,
                        files=files,
                    )
                )
            else:
                failed_results.append(
                    Result.create_from(
                        name="Parse failure error",
                        info="All log parsing attempts failed",
                        status=Result.Status.FAIL,
                    )
                )

    if server_died and not failed_results:
        failed_results.append(
            Result.create_from(
                name="Server died",
                info="Server died and no specific error was extracted",
                status=Result.Status.FAIL,
            )
        )

    if exit_code != 0:
        failed_results.append(
            Result.create_from(
                name="Check failed",
                info=f"Check failed with exit code {exit_code}",
                status=Result.Status.FAIL,
            )
        )

    all_results = failed_results + [r for r in test_results if r.is_ok()]
    r = Result.create_from(
        results=all_results,
        status=Result.Status.OK if not failed_results else "",
        stopwatch=stopwatch,
    )
    if not r.is_ok() and is_oom:
        r.set_status(Result.Status.OK)
        r.set_info("OOM error (allowed in stress tests)")

    if r.is_ok() and exit_code != 0 and not is_oom:
        r.set_failed().set_info(
            f"Unknown error: Test script failed with exit code {exit_code}"
        )

    r.set_files(additional_logs).set_files(core_files).complete_job()


if __name__ == "__main__":
    run_stress_test()