diff --git a/jenkins/L0_Test.groovy b/jenkins/L0_Test.groovy index 66f9a27c2409..3fb087b0fa00 100644 --- a/jenkins/L0_Test.groovy +++ b/jenkins/L0_Test.groovy @@ -47,7 +47,6 @@ ARTIFACT_PATH = env.artifactPath ? env.artifactPath : "sw-tensorrt-generic/llm-a UPLOAD_PATH = env.uploadPath ? env.uploadPath : "sw-tensorrt-generic/llm-artifacts/${JOB_NAME}/${BUILD_NUMBER}" URM_ARTIFACTORY_BASE = "https://urm.nvidia.com/artifactory" ENABLE_UPLOAD_TEST_RESULTS = params.enableUploadTestResults != null ? params.enableUploadTestResults : true -ENABLE_S3_ECHO_STDOUT = params.enableS3EchoStdout != null ? params.enableS3EchoStdout : false X86_64_TRIPLE = "x86_64-linux-gnu" AARCH64_TRIPLE = "aarch64-linux-gnu" @@ -711,6 +710,7 @@ def isValidSlurmJobId(def slurmJobID) { def cleanUpSlurmResources(def pipeline, SlurmCluster cluster, String clusterName, String jobUID){ CloudManager.withSlurmFrontendFailover(pipeline, clusterName, cluster) { remote -> def jobWorkspace = "/home/svc_tensorrt/bloom/scripts/${jobUID}" + def s3SpoolRoot = "/home/svc_tensorrt/bloom/scripts/.s3-spool-${jobUID}" Utils.exec(pipeline, script: "echo Sleeping to allow Slurm job completion; sleep 30") @@ -746,7 +746,7 @@ def cleanUpSlurmResources(def pipeline, SlurmCluster cluster, String clusterName // instead of deleting per job; reused images keep a refreshed mtime. "find ${cluster.scratchPath}/users/svc_tensorrt/containers -maxdepth 1 -name 'container-*.sqsh' -mtime +3 -delete 2>/dev/null || true", "find ${cluster.scratchPath}/users/svc_tensorrt/containers -maxdepth 1 \\( -name 'container-*.tmp' -o -name 'container-*.lock' \\) -mtime +1 -delete 2>/dev/null || true", - "rm -rf ${jobWorkspace} || true", + "rm -rf ${jobWorkspace} ${s3SpoolRoot} || true", ].join(" ; ") Utils.exec( pipeline, @@ -1529,9 +1529,6 @@ def getPytestBaseCommandLine( } def unittestMarkExpr = (stageName.startsWith("CPU-")) ? "cpu_only" : "not cpu_only" testCmdLine += ["--unittest-markexpr='${unittestMarkExpr}'"] - if (ENABLE_UPLOAD_TEST_RESULTS) { - testCmdLine += ["-o console_output_style=progress-even-when-capture-no"] - } if (extraArgs) { testCmdLine += extraArgs } @@ -1796,17 +1793,12 @@ def runLLMTestlistWithSbatch(pipeline, platform, testList, config=VANILLA_CONFIG "--group $splitId", *clusterDurationsArgsNode, ] - if (ENABLE_UPLOAD_TEST_RESULTS) { + if (ENABLE_UPLOAD_TEST_RESULTS && !testFilter[(DETAILED_LOG)]) { extraArgs += [ - "-s", + "--capture=fd", "--s3-upload-path=${uploadPath}/${stageName}", + "--s3-upload-mode=deferred", ] - if (ENABLE_S3_ECHO_STDOUT) { - extraArgs += [ - "--s3-echo-stdout", - "--s3-capture-mode=timestamped", - ] - } } def pytestCommand = getPytestBaseCommandLine( llmSrcNode, @@ -4488,17 +4480,12 @@ def runLLMTestlistOnPlatformImpl(pipeline, platform, testList, config=VANILLA_CO // Temporarily disable to reduce the log size // sh 'if [ "$(id -u)" -eq 0 ]; then dmesg -C || true; fi' def extraArgs = [*clusterDurationsArgs] - if (ENABLE_UPLOAD_TEST_RESULTS) { + if (ENABLE_UPLOAD_TEST_RESULTS && !testFilter[(DETAILED_LOG)]) { extraArgs += [ - "-s", + "--capture=fd", "--s3-upload-path=${uploadPath}/${stageName}", + "--s3-upload-mode=deferred", ] - if (ENABLE_S3_ECHO_STDOUT) { - extraArgs += [ - "--s3-echo-stdout", - "--s3-capture-mode=timestamped", - ] - } } def pytestCommand = getPytestBaseCommandLine( llmSrc, @@ -4533,57 +4520,66 @@ def runLLMTestlistOnPlatformImpl(pipeline, platform, testList, config=VANILLA_CO ]) { sh "env | sort" try { - if (preprocessedLists.regularCount > 0) { - sh """ - rm -rf ${stageName}/ && \ - cd ${llmSrc}/tests/integration/defs && \ - ${pytestCommand.join(" ")} - """ - } else { - echo "No regular tests to run for stage ${stageName}" - noRegularTests = true - sh "mkdir -p ${stageName}" - // Create an empty results.xml file for consistency - sh """ - echo '' > ${stageName}/results.xml - echo '' >> ${stageName}/results.xml - echo '' >> ${stageName}/results.xml - echo '' >> ${stageName}/results.xml - echo '' >> ${stageName}/results.xml - """ - } - } catch (InterruptedException e) { - throw e - } catch (Exception e) { - def isRerunFailed = rerunFailedTests(stageName, llmSrc, pytestCommand, "results.xml", "regular") - if (isRerunFailed) { - catchError(buildResult: 'SUCCESS', stageResult: 'FAILURE') { - error "Regular tests failed after rerun attempt" + try { + if (preprocessedLists.regularCount > 0) { + sh """ + rm -rf ${stageName}/ && \ + cd ${llmSrc}/tests/integration/defs && \ + ${pytestCommand.join(" ")} + """ + } else { + echo "No regular tests to run for stage ${stageName}" + noRegularTests = true + sh "mkdir -p ${stageName}" + // Create an empty results.xml file for consistency + sh """ + echo '' > ${stageName}/results.xml + echo '' >> ${stageName}/results.xml + echo '' >> ${stageName}/results.xml + echo '' >> ${stageName}/results.xml + echo '' >> ${stageName}/results.xml + """ } - rerunFailed = true - } else if (generateTimeoutTestResultXml(pipeline, stageName)) { - // Rerun passed but the first run had a timeout: mark this - // stage FAILURE so "[${stageName}] Run Pytest" turns red, - // not just the enclosing parent stage. - catchError(buildResult: 'SUCCESS', stageResult: 'FAILURE') { - error "Some tests terminated unexpectedly, please check the test report." + } catch (InterruptedException e) { + throw e + } catch (Exception e) { + def isRerunFailed = rerunFailedTests(stageName, llmSrc, pytestCommand, "results.xml", "regular") + if (isRerunFailed) { + catchError(buildResult: 'SUCCESS', stageResult: 'FAILURE') { + error "Regular tests failed after rerun attempt" + } + rerunFailed = true + } else if (generateTimeoutTestResultXml(pipeline, stageName)) { + // Rerun passed but the first run had a timeout: mark this + // stage FAILURE so "[${stageName}] Run Pytest" turns red, + // not just the enclosing parent stage. + catchError(buildResult: 'SUCCESS', stageResult: 'FAILURE') { + error "Some tests terminated unexpectedly, please check the test report." + } } } - } - // Run the isolated tests if exists - if (preprocessedLists.isolateCount > 0) { - stage ("[${stageName}] Run Pytest (Isolated)") { - echo "There are ${preprocessedLists.isolateCount} isolated tests to run" - rerunFailed = runIsolatedTests(preprocessedLists, pytestCommand, llmSrc, stageName) || rerunFailed + // Run the isolated tests if exists + if (preprocessedLists.isolateCount > 0) { + stage ("[${stageName}] Run Pytest (Isolated)") { + echo "There are ${preprocessedLists.isolateCount} isolated tests to run" + rerunFailed = runIsolatedTests(preprocessedLists, pytestCommand, llmSrc, stageName) || rerunFailed + } + } else { + echo "No isolated tests to run for stage ${stageName}" + noIsolateTests = true } - } else { - echo "No isolated tests to run for stage ${stageName}" - noIsolateTests = true - } - if (noRegularTests && noIsolateTests) { - error "No tests were executed for stage ${stageName}, please check the test list and test-db rendering result." + if (noRegularTests && noIsolateTests) { + error "No tests were executed for stage ${stageName}, please check the test list and test-db rendering result." + } + } finally { + if (ENABLE_UPLOAD_TEST_RESULTS && !testFilter[(DETAILED_LOG)]) { + sh """ + python3 ${llmSrc}/tests/test_common/s3_output.py \ + --drain-spool "${WORKSPACE}/${stageName}" || true + """ + } } } diff --git a/jenkins/scripts/slurm_run.sh b/jenkins/scripts/slurm_run.sh index 25622daff391..4aecf0d98864 100755 --- a/jenkins/scripts/slurm_run.sh +++ b/jenkins/scripts/slurm_run.sh @@ -82,6 +82,8 @@ perf_report_exit_code=0 eval $pytestCommand pytest_exit_code=$? echo "Rank${SLURM_PROCID} Pytest finished execution with exit code $pytest_exit_code" +python3 "$llmSrcNode/tests/test_common/s3_output.py" \ + --drain-spool "$jobWorkspace" || true # DEBUG: Diagnose intermittent "unrecognized arguments" failure (Exit Code 4) # Remove this after the issue is resolved diff --git a/tests/integration/defs/pytest.ini b/tests/integration/defs/pytest.ini index 45793bd9f51c..f8090a8052d7 100644 --- a/tests/integration/defs/pytest.ini +++ b/tests/integration/defs/pytest.ini @@ -1,5 +1,7 @@ [pytest] asyncio_default_fixture_loop_scope = module +log_format = %(asctime)s.%(msecs)03d %(levelname)-8s %(name)s:%(filename)s:%(lineno)d %(message)s +log_date_format = %Y-%m-%d %H:%M:%S threadleak = True # Thread-\d+ \(_manager_spawn\) / session-reuse-* / session-prefetch-* belong to # a pool cached for reuse by the NEXT test (tests/test_common/session_reuse.py) diff --git a/tests/integration/defs/test_unittests.py b/tests/integration/defs/test_unittests.py index 471a6abe821a..565bd49b8c0e 100644 --- a/tests/integration/defs/test_unittests.py +++ b/tests/integration/defs/test_unittests.py @@ -180,12 +180,15 @@ def test_unittests_v2(llm_root, llm_venv, case: str, output_dir, request): command += ["--run-ray"] s3_secret_key = None + s3_output_module = None s3_upload_path = request.config.getoption("--s3-upload-path", default=None) if s3_upload_path: + from test_common import s3_output as s3_output_module + inner_output_dir = os.path.join(output_dir, "inner-s3", case_fn) inner_upload_path = os.path.join(s3_upload_path, "inner", case_fn) command += [ - "-s", + "--capture=fd", f"--output-dir={inner_output_dir}", f"--s3-upload-path={inner_upload_path}", "--s3-upload-mode=deferred", @@ -262,6 +265,18 @@ def run_command(cmd, num_workers=1): ) print(f"{'='*60}\n") return False + finally: + if s3_output_module is not None: + try: + drained = s3_output_module.drain_pending_uploads( + inner_output_dir, + secret_key=s3_secret_key, + ) + except Exception as e: + print(f"Failed to drain pending S3 test logs: {e}") + else: + if not drained: + print("Some pending S3 test logs could not be drained") return True if num_workers == 1: diff --git a/tests/test_common/s3_output.py b/tests/test_common/s3_output.py index 4f3f8f272770..22abda8a8b25 100644 --- a/tests/test_common/s3_output.py +++ b/tests/test_common/s3_output.py @@ -13,942 +13,719 @@ # See the License for the specific language governing permissions and # limitations under the License. import argparse -import io +import fcntl +import hashlib +import json import logging import os +import posixpath import re +import socket import sys -import threading import time -from concurrent.futures import ThreadPoolExecutor, as_completed +from concurrent.futures import FIRST_COMPLETED, Future, ThreadPoolExecutor, as_completed, wait from dataclasses import dataclass -from datetime import datetime +from pathlib import Path import pytest -from _pytest.logging import catching_logs logger = logging.getLogger(__name__) +_CAPTURE_SECTION_PATTERN = re.compile(r"^Captured (stdout|stderr|log)(?: (setup|call|teardown))?$") +_STREAM_FILENAMES = { + "stdout": "stdout", + "stderr": "stderr", + "log": "logging", +} +_SPOOL_ROOT_NAME = ".s3-spool" +_SPOOL_CONFIG_NAME = "upload-config.json" +_SPOOL_WRITE_CHARS = 1024 * 1024 +_FAILED_OUTPUT_MAX_LINES = 200 +_FAILED_OUTPUT_MAX_BYTES = 65536 + + +def _spool_root(output_path: str) -> Path: + """Keep spool files beside output_path so result archives do not include them. + + This avoids packing thousands of per-test files into results-*.tar.gz, but + requires callers to drain or clean the sibling directory explicitly. + """ + output_root = Path(os.path.abspath(output_path)) + return output_root.parent / f"{_SPOOL_ROOT_NAME}-{output_root.name}" + class EnvDefault(argparse.Action): def __init__(self, envvar, required=True, default=None, **kwargs): - if envvar: - if envvar in os.environ: - default = os.environ[envvar] + if envvar and envvar in os.environ: + default = os.environ[envvar] if required and default: required = False - super(EnvDefault, self).__init__(default=default, required=required, **kwargs) + super().__init__(default=default, required=required, **kwargs) def __call__(self, parser, namespace, values, option_string=None): setattr(namespace, self.dest, values) @dataclass(frozen=True) -class FileSlice: - path: str - offset: int - size: int +class PendingUpload: + source_path: str + object_key: str + test_name: str + filename: str -class FileSliceReader(io.RawIOBase): - def __init__(self, file_slice): - self._slice = file_slice - self._file = open(file_slice.path, "rb", buffering=0) - self._position = 0 - self._file.seek(file_slice.offset) +@dataclass(frozen=True) +class CapturedSection: + content: str + attempt: int + + +@dataclass +class CapturedStream: + test_name: str + filename: str + source_path: str + filesize: int = 0 + message: str = "" + force_output: bool = False + finalized: bool = False + + +def _create_s3_client( + endpoint_url: str, + aws_access_key_id: str, + aws_secret_access_key: str, +): + try: + import boto3 + except ModuleNotFoundError as exc: + raise RuntimeError("boto3 is required to upload test logs") from exc + + return boto3.client( + "s3", + endpoint_url=endpoint_url, + aws_access_key_id=aws_access_key_id, + aws_secret_access_key=aws_secret_access_key, + ) - def readable(self): - return True - def seekable(self): +def _process_is_alive(pid: int) -> bool: + try: + os.kill(pid, 0) + except ProcessLookupError: + return False + except PermissionError: return True + return True - def tell(self): - return self._position - - def seek(self, offset, whence=os.SEEK_SET): - if whence == os.SEEK_SET: - position = offset - elif whence == os.SEEK_CUR: - position = self._position + offset - elif whence == os.SEEK_END: - position = self._slice.size + offset - else: - raise ValueError(f"Unsupported whence: {whence}") - if position < 0: - raise ValueError("Negative seek position") - self._file.seek(self._slice.offset + position) - self._position = position - return position - - def readinto(self, buffer): - remaining = self._slice.size - self._position - if remaining <= 0: - return 0 - view = memoryview(buffer)[: min(len(buffer), remaining)] - read_size = self._file.readinto(view) - if read_size is None: - return 0 - self._position += read_size - return read_size - - def close(self): - if not self.closed: - self._file.close() - super().close() - - -class SessionFDSpool: - def __init__(self, target_fd, path): - self.target_fd = target_fd - self.path = path - self._saved_fd = None - self._spool_fd = None - self._attached = False - - def _flush_target_stream(self): - stream = sys.stdout if self.target_fd == 1 else sys.stderr - try: - stream.flush() - except (OSError, ValueError): - pass - def start(self): - self._flush_target_stream() - self._saved_fd = os.dup(self.target_fd) +def _remove_empty_parents(path: Path, stop: Path) -> None: + parent = path.parent + while parent != stop.parent: try: - self._spool_fd = os.open( - self.path, - os.O_WRONLY | os.O_CREAT | os.O_TRUNC | os.O_APPEND, - 0o600, - ) - os.dup2(self._spool_fd, self.target_fd, inheritable=True) - self._attached = True + parent.rmdir() except OSError: - if self._spool_fd is not None: - os.close(self._spool_fd) - self._spool_fd = None - os.close(self._saved_fd) - self._saved_fd = None - raise - - def suspend_parent(self): - if not self._attached: - return - self._flush_target_stream() - os.dup2(self._saved_fd, self.target_fd, inheritable=True) - self._attached = False - - def resume_parent(self): - if self._attached or self._spool_fd is None: - return - self._flush_target_stream() - os.dup2(self._spool_fd, self.target_fd, inheritable=True) - self._attached = True - - def snapshot(self): - self._flush_target_stream() - return os.fstat(self._spool_fd).st_size - - def stop(self): - self.suspend_parent() - if self._spool_fd is not None: - os.close(self._spool_fd) - self._spool_fd = None - if self._saved_fd is not None: - os.close(self._saved_fd) - self._saved_fd = None - - -class SessionCapture: - def __init__(self, output_path): - spool_dir = os.path.join(output_path, ".s3-spool") - os.makedirs(spool_dir, exist_ok=True) - suffix = f"{os.getpid()}-{time.time_ns()}" - self._spools = { - "stdout.log": SessionFDSpool(1, os.path.join(spool_dir, f"stdout-{suffix}.log")), - "stderr.log": SessionFDSpool(2, os.path.join(spool_dir, f"stderr-{suffix}.log")), - } - self._suspend_depth = 0 - self._started = False - - def start(self): - started = [] - try: - for spool in self._spools.values(): - spool.start() - started.append(spool) - except Exception: - for spool in reversed(started): - spool.stop() - raise - self._started = True - - def snapshot(self): - return {filename: spool.snapshot() for filename, spool in self._spools.items()} - - def slices_since(self, offsets): - current = self.snapshot() - return { - filename: FileSlice( - path=spool.path, - offset=offsets[filename], - size=max(0, current[filename] - offsets[filename]), - ) - for filename, spool in self._spools.items() - } - - def suspend_parent(self): - if not self._started: - return - self._suspend_depth += 1 - if self._suspend_depth == 1: - for spool in self._spools.values(): - spool.suspend_parent() - - def resume_parent(self): - if not self._started or self._suspend_depth == 0: return - self._suspend_depth -= 1 - if self._suspend_depth == 0: - for spool in self._spools.values(): - spool.resume_parent() - - def stop(self): - if not self._started: + if parent == stop: return - self._suspend_depth = 0 - for spool in self._spools.values(): - spool.stop() - self._started = False - - def remove_files(self): - for spool in self._spools.values(): - try: - os.remove(spool.path) - except FileNotFoundError: - pass - try: - os.rmdir(os.path.dirname(next(iter(self._spools.values())).path)) - except OSError: - pass - - -class FDRedirector: - def __init__( - self, - target_fd, - log_file_path, - echo_to_original=False, - timestamp_format="%(asctime)s.%(msecs)03d", - date_format="%Y-%m-%d %H:%M:%S", - ): - self.target_fd = target_fd - self.log_file_path = log_file_path - self.echo_to_original = echo_to_original - self.timestamp_format = timestamp_format - self.date_format = date_format - - self.saved_fd = None - self._reader_thread = None - - def _flush_target_stream(self): - streams = [] - if self.target_fd == 1: - streams = [sys.stdout] - elif self.target_fd == 2: - streams = [sys.stderr] - else: - streams = [sys.stdout, sys.stderr] + parent = parent.parent - for stream in streams: - try: - stream.flush() - except Exception: - pass - - def __enter__(self): - # Drain any pending buffered writes to the current target fd before we - # redirect it. sys.stdout/sys.stderr are block-buffered when connected - # to a pipe; without this, late flushes (e.g. of a previous test's - # post-teardown print) would land in this test's file after we swap - # the underlying fd. - for stream in (sys.stdout, sys.stderr): - try: - stream.flush() - except Exception: - pass - - log_file = open(self.log_file_path, "w", encoding="utf-8", buffering=1) - pipe_read, pipe_write = os.pipe() - self.saved_fd = os.dup(self.target_fd) - os.dup2(pipe_write, self.target_fd) - os.close(pipe_write) - pipe_stream = os.fdopen(pipe_read, "r", encoding="utf-8", errors="replace", buffering=1) - - # Child processes may inherit the redirected fd and keep the pipe open - # after capture is restored. Do not let that block pytest shutdown. - self._reader_thread = threading.Thread( - target=self._reader_loop, args=(pipe_stream, log_file), daemon=True - ) - self._reader_thread.start() - - return self - - def _reader_loop(self, pipe_stream, log_file): - need_timestamp = True +def drain_pending_uploads(output_path: str, secret_key: str | None = None) -> bool: + """Upload files left by pytest processes that exited before session finish.""" + spool_root = _spool_root(output_path) + if not spool_root.exists(): + return True - try: - while True: - chunk = pipe_stream.read(4096) - if not chunk: - break - - # Build the timestamp prefix once per chunk and reuse it for every - # line in that chunk. The previous implementation walked the chunk - # char-by-char in Python, which was the bottleneck under high-volume - # output (e.g. tqdm progress bars, repetitive kernel info logs). - # Within a single 4 KiB chunk the timestamp would have been - # identical anyway (the reader processes the chunk under the GIL), - # so reusing one prefix is semantics-preserving. - now = datetime.now() - ts_str = self.timestamp_format % { - "asctime": now.strftime(self.date_format), - "msecs": now.microsecond // 1000, - } - prefix = f"[{ts_str}] " - - parts = [] - for line in chunk.splitlines(keepends=True): - if need_timestamp: - parts.append(prefix) - parts.append(line) - need_timestamp = line.endswith("\n") - log_file.write("".join(parts)) - log_file.flush() - - if self.echo_to_original and self.saved_fd is not None: - ret = os.write(self.saved_fd, chunk.encode("utf-8")) - if ret != len(chunk): - logger.warning( - f"Partial write to original FD {self.target_fd}: {ret} != {len(chunk)}" - ) - - except Exception as e: - logger.error(f"Error reading from pipe: {e}") - finally: - log_file.close() - pipe_stream.close() - - def __exit__(self, exc_type, exc_val, exc_tb): - if self.saved_fd is not None: - self._flush_target_stream() - # Restore the original fd binding. This also closes the previous - # binding (our pipe's write end), so the reader thread will see - # EOF and finish draining. - os.dup2(self.saved_fd, self.target_fd) - # NB: keep self.saved_fd valid until the reader thread joins; - # the reader may still drain pending bytes and want to echo them - # through saved_fd. Closing it here would race with that write. - - if self._reader_thread: - self._reader_thread.join(timeout=5.0) - if self._reader_thread.is_alive(): - logger.warning(f"Reader thread for FD {self.target_fd} did not exit in time") - - if self.saved_fd is not None: - os.close(self.saved_fd) - self.saved_fd = None + config_paths = list(spool_root.rglob(_SPOOL_CONFIG_NAME)) + if not config_paths: + return True + secret_key = secret_key or os.environ.get("S3_SECRET_KEY") + if not secret_key: + logger.warning("Cannot drain S3 test logs without S3_SECRET_KEY") return False + success = True + current_host = socket.gethostname() + for config_path in config_paths: + try: + with config_path.open("r", encoding="utf-8") as config_file: + try: + fcntl.flock(config_file, fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError: + continue + + config = json.load(config_file) + owner_host = config.get("hostname") + owner_pid = int(config.get("pid", 0)) + # PID reuse may leave a stale spool stranded. That short race is + # preferable to uploading files still being written. + if owner_host != current_host or (owner_pid and _process_is_alive(owner_pid)): + continue + + spool_dir = config_path.parent + source_paths = [ + path for path in spool_dir.rglob("*") if path.is_file() and path != config_path + ] + config_success = True + if source_paths: + client = _create_s3_client( + config["endpoint_url"], + config["aws_access_key_id"], + secret_key, + ) + uploads = {} + worker_count = min( + max(1, int(config.get("upload_workers", 8))), + len(source_paths), + ) + with ThreadPoolExecutor(max_workers=worker_count) as executor: + for source_path in source_paths: + relative_path = source_path.relative_to(spool_dir) + object_key = posixpath.join(config["upload_path"], *relative_path.parts) + future = executor.submit( + client.upload_file, + str(source_path), + config["bucket"], + object_key, + ExtraArgs={"ContentType": "text/plain"}, + ) + uploads[future] = (source_path, object_key) + + for future in as_completed(uploads): + source_path, object_key = uploads[future] + try: + future.result() + except Exception as exc: + config_success = False + success = False + logger.warning( + "Failed to drain S3 test log %s to %s: %s", + source_path, + object_key, + exc, + ) + else: + source_path.unlink(missing_ok=True) + _remove_empty_parents(source_path, spool_dir) + + if config_success: + config_path.unlink(missing_ok=True) + _remove_empty_parents(config_path, spool_dir) + try: + spool_dir.parent.rmdir() + except OSError: + pass + except FileNotFoundError: + # Another rank may have drained and removed this config after rglob. + continue + except (OSError, ValueError, KeyError, json.JSONDecodeError) as exc: + success = False + logger.warning("Failed to read S3 spool config %s: %s", config_path, exc) -class DirectFDRedirector: - def __init__(self, target_fd, log_file_path): - self.target_fd = target_fd - self.log_file_path = log_file_path - - self.saved_fd = None - self._log_file = None - - def _flush_target_stream(self): - streams = [] - if self.target_fd == 1: - streams = [sys.stdout] - elif self.target_fd == 2: - streams = [sys.stderr] - else: - streams = [sys.stdout, sys.stderr] - - for stream in streams: - try: - stream.flush() - except Exception: - pass - - def __enter__(self): - for stream in (sys.stdout, sys.stderr): - try: - stream.flush() - except Exception: - pass - - self._log_file = open(self.log_file_path, "w", encoding="utf-8", buffering=1) - self.saved_fd = os.dup(self.target_fd) - os.dup2(self._log_file.fileno(), self.target_fd) - return self - - def __exit__(self, exc_type, exc_val, exc_tb): - self._flush_target_stream() - - if self.saved_fd is not None: - os.dup2(self.saved_fd, self.target_fd) - os.close(self.saved_fd) - self.saved_fd = None - - if self._log_file is not None: - self._log_file.close() - self._log_file = None - - return False + return success class UploadLogPlugin: def __init__( self, - endpoint_url, - aws_access_key_id, - aws_secret_access_key, - bucket, - upload_path, - output_path, - echo_to_stdout=False, - skip_upload=False, - capture_mode="session", - upload_mode="sync", - upload_workers=8, - inline_output_max_bytes=256, - session_capture=None, - ): - self.upload_path = upload_path - self.output_path = output_path - self.bucket = bucket + endpoint_url: str, + aws_access_key_id: str, + aws_secret_access_key: str | None, + bucket: str, + upload_path: str, + output_path: str, + skip_upload: bool = False, + upload_mode: str = "sync", + upload_workers: int = 8, + inline_output_max_bytes: int = 256, + ) -> None: self.endpoint_url = endpoint_url self.aws_access_key_id = aws_access_key_id - self.echo_to_stdout = echo_to_stdout + self.bucket = bucket + self.upload_path = upload_path + self.output_path = output_path self.skip_upload = skip_upload - self.capture_mode = capture_mode self.upload_mode = upload_mode self.upload_workers = max(1, upload_workers) self.inline_output_max_bytes = inline_output_max_bytes if self.inline_output_max_bytes < 0: raise ValueError("--s3-inline-output-max-bytes must be >= 0") - if self.capture_mode in ("session", "direct") and self.echo_to_stdout: - raise ValueError( - f"--s3-capture-mode={self.capture_mode} cannot be used with --s3-echo-stdout" - ) + if self.upload_mode not in ("sync", "deferred"): + raise ValueError("--s3-upload-mode must be 'sync' or 'deferred'") + self.s3 = None if not self.skip_upload: - try: - import boto3 - except ModuleNotFoundError as exc: - raise RuntimeError( - "boto3 is required when --s3-upload-path is set without --s3-skip-upload" - ) from exc - self.s3 = boto3.client( - "s3", - endpoint_url=endpoint_url, - aws_access_key_id=aws_access_key_id, - aws_secret_access_key=aws_secret_access_key, + if not aws_secret_access_key: + raise ValueError("S3 secret key is required to upload test logs") + self.s3 = _create_s3_client( + endpoint_url, + aws_access_key_id, + aws_secret_access_key, ) - # nodeid -> dict of open capture state + logging handler. Session mode - # spans setup, call, and teardown; legacy modes close after the call. - self._active_capture: dict = {} - self._test_names: dict = {} - self._deferred_uploads = [] - self._captured_slices = {} - self._session_capture = session_capture - self._upload_failed = False - def normalize_test_name(self, nodeid): - import hashlib + suffix = f"{socket.gethostname()}-{os.getpid()}-{time.time_ns()}" + self._spool_dir = str(_spool_root(output_path) / suffix) + self._spool_config_path = os.path.join(self._spool_dir, _SPOOL_CONFIG_NAME) + self._test_names: dict[str, str] = {} + self._used_test_names: set[str] = set() + self._captured_sections: dict[str, dict[tuple[str, str, int], CapturedSection]] = {} + self._captured_streams: dict[str, dict[tuple[int, str], CapturedStream]] = {} + self._pending_reruns: set[str] = set() + self._upload_failed = False + self._executor = None + self._pending_uploads: dict[Future, PendingUpload] = {} + self._max_pending_uploads = self.upload_workers * 2 + if not self.skip_upload: + self._write_spool_config() + if self.upload_mode == "deferred": + self._executor = ThreadPoolExecutor( + max_workers=self.upload_workers, + thread_name_prefix="s3-test-log-upload", + ) + def normalize_test_name(self, nodeid: str) -> str: test_name = re.sub(r"[^\w\-]", "_", nodeid) - suffix = hashlib.md5(nodeid.encode()).hexdigest()[:8] + suffix = hashlib.md5(nodeid.encode(), usedforsecurity=False).hexdigest()[:8] timestamp = int(time.time()) - # Linux limits a single path component to 255 bytes. if len(test_name) > 200: test_name = test_name[:200] return f"{test_name}-{suffix}-{timestamp}" - def _open_capture(self, item): - """Open stdout/stderr and logging capture for ``item``. - - Returns a state dict on success, or ``None`` if setup failed (in which - case the test runs uncaptured). Never propagates exceptions. - """ - state = {} - try: - test_name = self.normalize_test_name(item.nodeid) - state["test_name"] = test_name - output_path = os.path.join(self.output_path, test_name) - os.makedirs(output_path, exist_ok=True) - - log_file = os.path.join(output_path, "logging.log") - - log_date_format = item.config.getini("log_date_format") - log_format = item.config.getini("log_format") - - timestamp_format = None - if log_format: - match = re.search(r"\[([^\]]*%\(asctime\)s[^\]]*)\]", log_format) - if match: - timestamp_format = match.group(1) - - handler = logging.FileHandler(log_file) - logging_plugin = item.config.pluginmanager.getplugin("logging-plugin") - if logging_plugin is not None: - handler.setFormatter(logging_plugin.formatter) - elif log_format: - formatter = logging.Formatter(log_format, datefmt=log_date_format) - handler.setFormatter(formatter) - state["handler"] = handler - - if self.capture_mode == "session": - if self._session_capture is None: - raise RuntimeError("Session capture has not started") - state["session_offsets"] = self._session_capture.snapshot() - else: - stdout_file = os.path.join(output_path, "stdout.log") - stderr_file = os.path.join(output_path, "stderr.log") - fd_kwargs = {} - if log_date_format: - fd_kwargs["date_format"] = log_date_format - if timestamp_format: - fd_kwargs["timestamp_format"] = timestamp_format - - if self.capture_mode == "direct": - state["stdout_redir"] = DirectFDRedirector(1, stdout_file) - else: - state["stdout_redir"] = FDRedirector( - 1, stdout_file, echo_to_original=self.echo_to_stdout, **fd_kwargs - ) - state["stdout_redir"].__enter__() - if self.capture_mode == "direct": - state["stderr_redir"] = DirectFDRedirector(2, stderr_file) - else: - state["stderr_redir"] = FDRedirector( - 2, stderr_file, echo_to_original=self.echo_to_stdout, **fd_kwargs - ) - state["stderr_redir"].__enter__() - state["log_cm"] = catching_logs(handler) - state["log_cm"].__enter__() - return state - except Exception as e: - logger.warning( - "S3 capture setup failed for %r: %s; running without capture", item.nodeid, e - ) - self._close_capture(state) - return None - - def _close_capture(self, state): - """Close capture state opened by ``_open_capture``. Never raises.""" - if not state: - return - session_offsets = state.get("session_offsets") - if session_offsets is not None and self._session_capture is not None: - try: - slices = self._session_capture.slices_since(session_offsets) - for filename, file_slice in slices.items(): - self._captured_slices[(state["test_name"], filename)] = file_slice - except (OSError, ValueError) as e: - logger.warning("Error finalizing session capture: %s", e) - # Close in reverse order so fd 1/2 are restored before logging stops. - for key in ("log_cm", "stderr_redir", "stdout_redir"): - cm = state.get(key) - if cm is None: - continue - try: - cm.__exit__(None, None, None) - except Exception as e: - logger.warning("Error closing %s capture: %s", key, e) - handler = state.get("handler") - if handler is not None: - try: - handler.close() - except Exception: - pass - - @pytest.hookimpl(wrapper=True) - def pytest_runtest_setup(self, item): - state = self._open_capture(item) - if state is not None: - self._active_capture[item.nodeid] = state - self._test_names[item.nodeid] = state["test_name"] - yield - - @pytest.hookimpl(wrapper=True) - def pytest_sessionstart(self, session): - result = yield - if self.capture_mode == "session": - if self._session_capture is None: - self._session_capture = SessionCapture(self.output_path) - self._session_capture.start() - else: - self._session_capture.resume_parent() - return result - - @pytest.hookimpl(wrapper=True) - def pytest_runtest_makereport(self, item, call): - report = yield - if self.capture_mode != "session" and ( - report.when == "call" or (report.when == "setup" and report.outcome != "passed") - ): - self._close_capture(self._active_capture.pop(item.nodeid, None)) - return report - - @pytest.hookimpl(wrapper=True) - def pytest_runtest_teardown(self, item, nextitem): - if self.capture_mode == "session": - try: - return (yield) - finally: - self._close_capture(self._active_capture.pop(item.nodeid, None)) - - # Fallback for custom or interrupted runtest protocols that did not - # produce a call report. Normal execution closes capture earlier. - state = self._active_capture.pop(item.nodeid, None) - self._close_capture(state) - if state is not None: - logger.warning( - "S3 capture for %r remained active until teardown; " - "pytest result/progress output may have been captured", - item.nodeid, - ) - yield - - def _suspend_session_capture(self): - if self._session_capture is not None: - self._session_capture.suspend_parent() + def _write_spool_config(self) -> None: + os.makedirs(self._spool_dir, exist_ok=True) + config = { + "endpoint_url": self.endpoint_url, + "aws_access_key_id": self.aws_access_key_id, + "bucket": self.bucket, + "upload_path": self.upload_path, + "hostname": socket.gethostname(), + "pid": os.getpid(), + "upload_workers": self.upload_workers, + } + temporary_path = f"{self._spool_config_path}.{os.getpid()}.tmp" + with open(temporary_path, "w", encoding="utf-8") as config_file: + json.dump(config, config_file) + os.replace(temporary_path, self._spool_config_path) + + def _test_name(self, nodeid: str) -> str: + test_name = self._test_names.get(nodeid) + if test_name is None: + test_name = self.normalize_test_name(nodeid) + base_name = test_name + collision_index = 1 + while test_name in self._used_test_names: + test_name = f"{base_name}-{collision_index}" + collision_index += 1 + self._used_test_names.add(test_name) + self._test_names[nodeid] = test_name + return test_name + + def _object_key(self, test_name: str, filename: str) -> str: + return posixpath.join(self.upload_path, test_name, filename) + + def _file_url(self, object_key: str) -> str: + return posixpath.join( + self.endpoint_url, + "v1/AUTH_" + self.aws_access_key_id, + self.bucket, + object_key, + ) - def _resume_session_capture(self): - if self._session_capture is not None: - self._session_capture.resume_parent() + def _append_spool_file(self, test_name: str, filename: str, content: str) -> str: + if not self.skip_upload and not os.path.exists(self._spool_config_path): + self._write_spool_config() + test_dir = os.path.join(self._spool_dir, test_name) + os.makedirs(test_dir, exist_ok=True) + source_path = os.path.join(test_dir, filename) + file_descriptor = os.open( + source_path, + os.O_WRONLY | os.O_CREAT | os.O_APPEND, + 0o600, + ) + with os.fdopen(file_descriptor, "a", encoding="utf-8", errors="replace") as output: + for offset in range(0, len(content), _SPOOL_WRITE_CHARS): + output.write(content[offset : offset + _SPOOL_WRITE_CHARS]) + return source_path - @pytest.hookimpl(wrapper=True) - def pytest_runtest_logstart(self, nodeid, location): - self._suspend_session_capture() + def _remove_source(self, source_path: str) -> None: + path = Path(source_path) try: - return (yield) - finally: - self._resume_session_capture() - - @pytest.hookimpl(wrapper=True) - def pytest_runtest_logfinish(self, nodeid, location): - self._suspend_session_capture() + path.unlink() + except FileNotFoundError: + return + except OSError as exc: + logger.warning("Failed to remove local S3 log file %s: %s", source_path, exc) + return + _remove_empty_parents(path, Path(self._spool_dir)) try: - return (yield) - finally: - self._resume_session_capture() + Path(self._spool_dir).parent.rmdir() + except OSError: + pass - def get_file_size(self, path): - try: - return os.path.getsize(path) - except FileNotFoundError: - return None - - def _capture_source(self, test_name, filename): - file_slice = self._captured_slices.pop((test_name, filename), None) - if file_slice is not None: - return file_slice - return os.path.join(self.output_path, test_name, filename) - - def _source_exists(self, source): - if isinstance(source, FileSlice): - return os.path.exists(source.path) - return os.path.exists(source) - - def _source_size(self, source): - if isinstance(source, FileSlice): - return source.size - return os.path.getsize(source) - - def _open_source(self, source): - if isinstance(source, FileSlice): - return io.BufferedReader(FileSliceReader(source)) - return open(source, "rb") - - def _remove_source(self, source): - if not isinstance(source, FileSlice): - self._remove_local_log_file(source) - - def _object_key(self, test_name, filename): - return os.path.join(self.upload_path, test_name, filename) - - def _file_url(self, object_key): - return os.path.join( - self.endpoint_url, - "v1/AUTH_" + self.aws_access_key_id, + def _upload_source(self, source_path: str, object_key: str) -> None: + self.s3.upload_file( + source_path, self.bucket, object_key, + ExtraArgs={"ContentType": "text/plain"}, ) - def _upload_source(self, source, object_key): - extra_args = {"ContentType": "text/plain"} - if isinstance(source, FileSlice): - with self._open_source(source) as source_file: - self.s3.upload_fileobj( - source_file, - self.bucket, - object_key, - ExtraArgs=extra_args, + def _finish_uploads(self, futures: set[Future]) -> None: + for future in futures: + upload = self._pending_uploads.pop(future) + try: + future.result() + except Exception as exc: + self._upload_failed = True + logger.warning( + "Deferred upload failed. test_name: %s, filename: %s, " + "object_key: %s, source: %s, error: %s", + upload.test_name, + upload.filename, + upload.object_key, + upload.source_path, + exc, ) - else: - self.s3.upload_file( - source, - self.bucket, - object_key, - ExtraArgs=extra_args, - ) + else: + self._remove_source(upload.source_path) - def _append_upload_failed(self, report, section_name, source, filesize, error): - with self._open_source(source) as source_file: - limit = 65536 - # Limit content to 64k (65536 bytes) - trail_content = "... [truncated]" - content = source_file.read(limit + 1).decode("utf-8", errors="replace") - if len(content) > limit: - content = content[: limit - len(trail_content)] + trail_content - report.sections.append( - ( - section_name, - f"""upload failed: {error}\nsize: {filesize} bytes\ncontent: {content}""", - ) + def _reap_uploads(self, block_when_full: bool = False) -> None: + completed = {future for future in self._pending_uploads if future.done()} + if completed: + self._finish_uploads(completed) + if block_when_full and len(self._pending_uploads) >= self._max_pending_uploads: + completed, _ = wait(self._pending_uploads, return_when=FIRST_COMPLETED) + self._finish_uploads(completed) + + def _schedule_upload( + self, + source_path: str, + object_key: str, + test_name: str, + filename: str, + ) -> None: + self._reap_uploads(block_when_full=True) + future = self._executor.submit(self._upload_source, source_path, object_key) + self._pending_uploads[future] = PendingUpload( + source_path=source_path, + object_key=object_key, + test_name=test_name, + filename=filename, ) - def _should_inline_output(self, filename, filesize): - return filename in ("stdout.log", "stderr.log") and filesize < self.inline_output_max_bytes + def _tail(self, source_path: str) -> str: + filesize = os.path.getsize(source_path) + read_size = min(filesize, _FAILED_OUTPUT_MAX_BYTES) + with open(source_path, "rb") as source: + source.seek(filesize - read_size) + tail = source.read(read_size).decode("utf-8", errors="replace") + + lines = tail.splitlines(keepends=True) + truncated = filesize > read_size + if len(lines) > _FAILED_OUTPUT_MAX_LINES: + lines = lines[-_FAILED_OUTPUT_MAX_LINES:] + truncated = True + tail = "".join(lines) + if truncated: + tail = "... [truncated]\n" + tail + return tail + + @staticmethod + def _format_report_content(message: str, tail: str | None = None) -> str: + if tail is None: + return f"{message}\n" + formatted = f"{message}\n\nLast {_FAILED_OUTPUT_MAX_LINES} lines:\n{tail}" + if not formatted.endswith("\n"): + formatted += "\n" + return formatted + + @staticmethod + def _stream_filename(stream_name: str, attempt: int) -> str: + filename = _STREAM_FILENAMES[stream_name] + if attempt > 1: + filename += f"-attempt-{attempt}" + return f"{filename}.log" + + def _stream_message(self, stream: CapturedStream) -> str: + object_key = self._object_key(stream.test_name, stream.filename) + file_url = self._file_url(object_key) + if self.skip_upload: + return f"{stream.filesize} bytes (upload skipped, would upload to {file_url})" + return f"{stream.filesize} bytes scheduled for upload to {file_url}" - def _append_inline_output(self, report, section_name, source): - with self._open_source(source) as source_file: - content = source_file.read().decode("utf-8", errors="replace") - report.sections.append((section_name, content)) + def _append_capture( + self, + nodeid: str, + attempt: int, + stream_name: str, + content: str, + ) -> bool: + streams = self._captured_streams.setdefault(nodeid, {}) + stream_key = (attempt, stream_name) + stream = streams.get(stream_key) + if stream is None: + test_name = self._test_name(nodeid) + filename = self._stream_filename(stream_name, attempt) + source_path = self._append_spool_file(test_name, filename, content) + stream = CapturedStream( + test_name=test_name, + filename=filename, + source_path=source_path, + ) + streams[stream_key] = stream + else: + if stream.finalized: + logger.warning( + "Ignoring captured output received after %s was finalized", + stream.filename, + ) + return False + self._append_spool_file(stream.test_name, stream.filename, content) - def _remove_local_log_file(self, filepath): - try: - os.remove(filepath) - except FileNotFoundError: - return - except OSError as e: - logger.warning("Failed to remove local S3 log file %s: %s", filepath, e) - return + stream.filesize = os.path.getsize(stream.source_path) + stream.message = self._stream_message(stream) + return True - output_root = os.path.abspath(self.output_path) - parent = os.path.abspath(os.path.dirname(filepath)) - while parent != output_root and os.path.commonpath([output_root, parent]) == output_root: - try: - os.rmdir(parent) - except OSError: - break - parent = os.path.dirname(parent) + def _should_inline_stream(self, stream_name: str, stream: CapturedStream) -> bool: + return ( + stream_name in ("stdout", "stderr") + and self.inline_output_max_bytes > 0 + and stream.filesize < self.inline_output_max_bytes + ) - def upload_and_report(self, report, test_name, filename, section_name): - source = self._capture_source(test_name, filename) - if not self._source_exists(source): - report.sections.append((section_name, "")) + def _finalize_stream(self, stream_name: str, stream: CapturedStream) -> None: + if stream.finalized: return - filesize = self._source_size(source) - if filesize == 0: - report.sections.append((section_name, "")) - self._remove_source(source) + stream.finalized = True + if self._should_inline_stream(stream_name, stream): + self._remove_source(stream.source_path) return - if self._should_inline_output(filename, filesize): - self._append_inline_output(report, section_name, source) - self._remove_source(source) - return - object_key = self._object_key(test_name, filename) - fileurl = self._file_url(object_key) + + object_key = self._object_key(stream.test_name, stream.filename) + file_url = self._file_url(object_key) if self.skip_upload: - # Experiment knob: skip the actual S3 upload to measure how much - # of this plugin's per-test overhead comes from boto3 network IO. - report.sections.append( - ( - section_name, - f"{filesize} bytes (upload skipped, would upload to {fileurl})", - ) - ) + stream.message = self._stream_message(stream) return if self.upload_mode == "deferred": - self._deferred_uploads.append((source, object_key, test_name, filename)) - report.sections.append( - ( - section_name, - f"{filesize} bytes scheduled for upload to {fileurl}", - ) + self._schedule_upload( + stream.source_path, + object_key, + stream.test_name, + stream.filename, ) + stream.message = f"{stream.filesize} bytes scheduled for upload to {file_url}" return + try: - self._upload_source(source, object_key) - report.sections.append( - ( - section_name, - f"{filesize} bytes uploaded to {fileurl}", - ) - ) - self._remove_source(source) - except Exception as e: + self._upload_source(stream.source_path, object_key) + except Exception as exc: self._upload_failed = True logger.warning( - f"Upload failed. test_name: {test_name}, filename: {filename}, error: {e}" + "Upload failed. test_name: %s, filename: %s, error: %s", + stream.test_name, + stream.filename, + exc, + ) + stream.message = f"upload failed: {exc}\nsize: {stream.filesize} bytes" + stream.force_output = True + return + + self._remove_source(stream.source_path) + stream.message = f"{stream.filesize} bytes uploaded to {file_url}" + + def _finalize_attempt(self, nodeid: str, attempt: int) -> None: + streams = self._captured_streams.get(nodeid, {}) + for (stream_attempt, stream_name), stream in streams.items(): + if stream_attempt == attempt: + self._finalize_stream(stream_name, stream) + + def _finalize_node(self, nodeid: str) -> None: + for (_, stream_name), stream in self._captured_streams.get(nodeid, {}).items(): + self._finalize_stream(stream_name, stream) + + @staticmethod + def _attempt(report) -> int: + rerun = getattr(report, "rerun", 0) + return rerun + 1 if isinstance(rerun, int) and rerun >= 0 else 1 + + def _process_report(self, report, default_phase: str) -> None: + nodeid = getattr(report, "nodeid", "unknown") + failed = getattr(report, "outcome", None) == "failed" + if getattr(report, "outcome", None) == "rerun": + self._pending_reruns.add(nodeid) + attempt = self._attempt(report) + captured_sections = self._captured_sections.setdefault(nodeid, {}) + duplicate_counts: dict[tuple[str, str], int] = {} + section_entries = [] + represented_streams = [] + represented_stream_set = set() + for section_name, content in report.sections: + match = _CAPTURE_SECTION_PATTERN.fullmatch(section_name) + if match is None: + section_entries.append((section_name, content, None)) + continue + stream_name = match.group(1) + phase = match.group(2) or default_phase + duplicate_key = (stream_name, phase) + duplicate_index = duplicate_counts.get(duplicate_key, 0) + duplicate_counts[duplicate_key] = duplicate_index + 1 + + section_key = (stream_name, phase, duplicate_index) + captured_section = captured_sections.get(section_key) + if captured_section is None or captured_section.content != content: + if not self._append_capture( + nodeid, + attempt, + stream_name, + content, + ): + continue + captured_section = CapturedSection(content, attempt) + captured_sections[section_key] = captured_section + + stream_key = (captured_section.attempt, stream_name) + section_entries.append((section_name, content, stream_key)) + if stream_key not in represented_stream_set: + represented_stream_set.add(stream_key) + represented_streams.append(stream_key) + + failure_tails = {} + streams = self._captured_streams.get(nodeid, {}) + if failed: + for stream_key in represented_streams: + stream_attempt, stream_name = stream_key + stream = streams[stream_key] + if stream_attempt == attempt and not self._should_inline_stream( + stream_name, stream + ): + failure_tails[stream_key] = self._tail(stream.source_path) + + if default_phase in ("teardown", "collect"): + self._finalize_attempt(nodeid, attempt) + + transformed_sections = [] + reported_streams = set() + for section_name, content, stream_key in section_entries: + if stream_key is None: + transformed_sections.append((section_name, content)) + continue + _, stream_name = stream_key + stream = streams[stream_key] + if self._should_inline_stream(stream_name, stream): + transformed_sections.append((section_name, content)) + continue + if stream_key in reported_streams: + continue + reported_streams.add(stream_key) + + tail = failure_tails.get(stream_key) + if tail is None and stream.force_output: + tail = self._tail(stream.source_path) + transformed_sections.append( + ( + f"Captured {stream_name}", + self._format_report_content(stream.message, tail), + ) ) - self._append_upload_failed(report, section_name, source, filesize, e) + report.sections[:] = transformed_sections + + def pytest_runtest_logstart(self, nodeid: str, location) -> None: + self._test_name(nodeid) - @pytest.hookimpl(wrapper=True) + @pytest.hookimpl(wrapper=True, tryfirst=True) def pytest_runtest_logreport(self, report): - self._suspend_session_capture() - try: - if report.when == "teardown": - test_name = self._test_names.pop(report.nodeid, None) - if test_name is None: - test_name = self.normalize_test_name(report.nodeid) - # Add S3 report sections before the junitxml plugin handles - # this report, while pytest's terminal output bypasses capture. - self.upload_and_report(report, test_name, "stdout.log", "Captured stdout") - self.upload_and_report(report, test_name, "stderr.log", "Captured stderr") - self.upload_and_report(report, test_name, "logging.log", "Captured log") - return (yield) - finally: - self._resume_session_capture() + self._process_report(report, getattr(report, "when", "call")) + return (yield) @pytest.hookimpl(tryfirst=True) - def pytest_sessionfinish(self, session, exitstatus): - for state in self._active_capture.values(): - self._close_capture(state) - self._active_capture.clear() - - if self._session_capture is not None: - self._session_capture.stop() - - if not self.skip_upload and self._deferred_uploads: - workers = min(self.upload_workers, len(self._deferred_uploads)) - logger.info( - "Uploading %d deferred S3 test log files with %d workers", - len(self._deferred_uploads), - workers, - ) - with ThreadPoolExecutor(max_workers=workers) as executor: - futures = { - executor.submit(self._upload_source, source, object_key): ( - source, - object_key, - test_name, - filename, - ) - for source, object_key, test_name, filename in self._deferred_uploads - } - for future in as_completed(futures): - source, object_key, test_name, filename = futures[future] - try: - future.result() - self._remove_source(source) - except Exception as e: - self._upload_failed = True - logger.warning( - "Deferred upload failed. test_name: %s, filename: %s, " - "object_key: %s, source: %s, error: %s", - test_name, - filename, - object_key, - source, - e, - ) - self._deferred_uploads.clear() - - if ( - self._session_capture is not None - and not self.skip_upload - and not self._upload_failed - and not self._captured_slices - ): - self._session_capture.remove_files() - - -def add_options(parser): - """Register S3 CLI options. Call from pytest_addoption in any conftest that needs S3 upload.""" + def pytest_collectreport(self, report) -> None: + self._process_report(report, "collect") + nodeid = getattr(report, "nodeid", "unknown") + self._test_names.pop(nodeid, None) + self._captured_sections.pop(nodeid, None) + self._captured_streams.pop(nodeid, None) + self._pending_reruns.discard(nodeid) + + def pytest_runtest_logfinish(self, nodeid: str, location) -> None: + self._finalize_node(nodeid) + self._test_names.pop(nodeid, None) + if nodeid in self._pending_reruns: + self._pending_reruns.remove(nodeid) + return + self._captured_sections.pop(nodeid, None) + self._captured_streams.pop(nodeid, None) + + @pytest.hookimpl(tryfirst=True) + def pytest_sessionfinish(self, session, exitstatus) -> None: + for nodeid in tuple(self._captured_streams): + self._finalize_node(nodeid) + if self._executor is not None: + self._executor.shutdown(wait=True) + self._executor = None + self._finish_uploads(set(self._pending_uploads)) + if not self.skip_upload and not self._upload_failed: + try: + Path(self._spool_config_path).unlink(missing_ok=True) + Path(self._spool_dir).rmdir() + Path(self._spool_dir).parent.rmdir() + except OSError: + pass + self._captured_sections.clear() + self._captured_streams.clear() + self._pending_reruns.clear() + + +def add_options(parser) -> None: + """Register S3 options on a pytest parser.""" parser.addoption( "--s3-endpoint", action=EnvDefault, envvar="S3_ENDPOINT", default="https://pbss.s8k.io", - help="s3 endpoint", + help="S3 endpoint", ) parser.addoption( "--s3-username", action=EnvDefault, envvar="S3_USERNAME", default="svc_tensorrt", - help="s3 username", + help="S3 username", ) parser.addoption( "--s3-secret-key", action=EnvDefault, envvar="S3_SECRET_KEY", required=False, - help="s3 secret key", + help="S3 secret key", ) parser.addoption( "--s3-bucket", action=EnvDefault, envvar="S3_BUCKET", default="trtllm-ci-logs", - help="s3 bucket name", + help="S3 bucket name", ) parser.addoption( "--s3-upload-path", action=EnvDefault, envvar="S3_UPLOAD_PATH", required=False, - help="s3 upload path", - ) - parser.addoption( - "--s3-echo-stdout", - action="store_true", - default=False, - help="Besides capturing stdout/stderr to per-test log files, also echo " - "them through to the original stdout/stderr for live debugging. " - "This requires --s3-capture-mode=timestamped and should be set on the outer pytest " - "invocation; nested pytest invocations spawned by individual tests " - "should NOT set this, to avoid duplicating their output back through " - "the outer pipe.", + help="S3 upload path", ) parser.addoption( "--s3-skip-upload", action="store_true", default=False, - help="Experiment knob: still capture stdout/stderr/log per test and " - "append URLs to report sections, but skip the actual s3.upload_file " - "call. Used to measure how much of the plugin's per-test overhead " - "comes from boto3/network IO vs. local capture machinery.", - ) - parser.addoption( - "--s3-capture-mode", - action="store", - choices=("session", "timestamped", "direct"), - default="session", - help="Capture each stream through a session-scoped append-only file and " - "split it into per-test ranges. Timestamped and direct retain the legacy " - "per-test FD redirection modes.", + help="Transform captured output into S3 report links without uploading it.", ) parser.addoption( "--s3-upload-mode", action="store", choices=("sync", "deferred"), default="sync", - help="Upload each per-test log file synchronously in the test teardown, " - "or defer uploads until pytest session finish. Deferred mode keeps " - "per-test files and report URLs while reducing teardown latency for " - "nested pytest runs with many cases.", + help="Upload report sections synchronously or with a bounded background pool.", ) parser.addoption( "--s3-upload-workers", @@ -962,42 +739,55 @@ def add_options(parser): action="store", type=int, default=256, - help="Inline captured stdout/stderr into the test report and skip S3 " - "upload when each file is smaller than this many bytes. Set to 0 to " - "disable inlining for non-empty output.", + help="Inline stdout/stderr smaller than this size instead of uploading it.", ) -def register_plugin(config, session_capture=None): - """Register UploadLogPlugin if --s3-upload-path and --output-dir are both set.""" +def register_plugin(config): + """Register the report transformer when S3 output is configured.""" + existing_plugin = config.pluginmanager.getplugin("upload_log_plugin") + if existing_plugin is not None: + return existing_plugin + s3_upload_path = config.getoption("--s3-upload-path", default=None) output_dir = config.getoption("--output-dir", default=None) if not (s3_upload_path and output_dir): return None - pytest_capture_mode = config.getoption("capture", default="no") - if pytest_capture_mode != "no": - raise ValueError("capture mode must be 'no' when upload path is specified") - s3_secret_key = config.getoption("--s3-secret-key") + capture_mode = config.getoption("capture", default="fd") + if capture_mode != "fd": + raise ValueError("--s3-upload-path requires pytest --capture=fd") + skip_upload = config.getoption("--s3-skip-upload", default=False) + s3_secret_key = config.getoption("--s3-secret-key", default=None) if not skip_upload and not s3_secret_key: raise ValueError( "--s3-secret-key (or S3_SECRET_KEY env var) is required when --s3-upload-path is set" ) - s3_capture_mode = config.getoption("--s3-capture-mode", default="session") + plugin = UploadLogPlugin( endpoint_url=config.getoption("--s3-endpoint"), aws_access_key_id=config.getoption("--s3-username"), aws_secret_access_key=s3_secret_key, bucket=config.getoption("--s3-bucket"), upload_path=s3_upload_path, - output_path=output_dir, - echo_to_stdout=config.getoption("--s3-echo-stdout", default=False), + output_path=os.path.abspath(output_dir), skip_upload=skip_upload, - capture_mode=s3_capture_mode, upload_mode=config.getoption("--s3-upload-mode", default="sync"), upload_workers=config.getoption("--s3-upload-workers", default=8), inline_output_max_bytes=config.getoption("--s3-inline-output-max-bytes", default=256), - session_capture=session_capture if s3_capture_mode == "session" else None, ) config.pluginmanager.register(plugin, "upload_log_plugin") return plugin + + +def _main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--drain-spool", metavar="OUTPUT_PATH") + args = parser.parse_args() + if args.drain_spool: + return 0 if drain_pending_uploads(args.drain_spool) else 1 + return 0 + + +if __name__ == "__main__": + sys.exit(_main()) diff --git a/tests/test_common/s3_output_hooks.py b/tests/test_common/s3_output_hooks.py index ccaf34f6ea02..39df46e8e932 100644 --- a/tests/test_common/s3_output_hooks.py +++ b/tests/test_common/s3_output_hooks.py @@ -12,31 +12,14 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. -"""Start S3 FD capture before pytest imports the initial conftests.""" +"""Register the S3 report transformer on pytest workers.""" -import argparse import os -from collections.abc import Generator -from dataclasses import dataclass -from typing import cast import pytest from test_common import s3_output -_CAPTURE_STATE_ATTRIBUTE = "_s3_output_early_capture_state" - - -@dataclass -class _EarlyCaptureState: - capture: s3_output.SessionCapture - claimed: bool = False - - def cleanup(self) -> None: - self.capture.stop() - if not self.claimed: - self.capture.remove_files() - def _is_xdist_worker(config: pytest.Config) -> bool: return bool(os.environ.get("PYTEST_XDIST_WORKER")) or hasattr(config, "workerinput") @@ -52,77 +35,7 @@ def _is_xdist_controller(config: pytest.Config) -> bool: return num_processes not in (None, 0, "0") -def _parse_early_capture_options( - early_config: pytest.Config, args: list[str] -) -> tuple[str | None, str | None, str, str]: - parser = argparse.ArgumentParser(add_help=False, allow_abbrev=False) - parser.add_argument("--output-dir", "-O") - parser.add_argument("--s3-upload-path") - parser.add_argument("--s3-capture-mode") - parsed, _ = parser.parse_known_args(args) - - namespace = early_config.known_args_namespace - output_path = cast( - str | None, - getattr(namespace, "output_dir", None) or parsed.output_dir, - ) - upload_path = cast( - str | None, - getattr(namespace, "s3_upload_path", None) - or parsed.s3_upload_path - or os.environ.get("S3_UPLOAD_PATH"), - ) - capture_mode = cast( - str, - getattr(namespace, "s3_capture_mode", None) or parsed.s3_capture_mode or "session", - ) - pytest_capture = cast(str, getattr(namespace, "capture", "fd")) - return output_path, upload_path, capture_mode, pytest_capture - - -def _capture_state(config: pytest.Config) -> _EarlyCaptureState | None: - return cast( - _EarlyCaptureState | None, - getattr(config, _CAPTURE_STATE_ATTRIBUTE, None), - ) - - -@pytest.hookimpl(wrapper=True, tryfirst=True) -def pytest_load_initial_conftests( - early_config: pytest.Config, args: list[str] -) -> Generator[None, object, object]: - output_path, upload_path, capture_mode, pytest_capture = _parse_early_capture_options( - early_config, args - ) - if ( - _capture_state(early_config) is not None - or _is_xdist_controller(early_config) - or not output_path - or not upload_path - or capture_mode != "session" - or pytest_capture != "no" - ): - return (yield) - - capture = s3_output.SessionCapture(output_path) - capture.start() - state = _EarlyCaptureState(capture) - setattr(early_config, _CAPTURE_STATE_ATTRIBUTE, state) - early_config.add_cleanup(state.cleanup) - try: - return (yield) - finally: - # MPI runtimes initialized while conftests load retain the spool FD. - # Restore the parent so pytest configuration output stays visible. - capture.suspend_parent() - - @pytest.hookimpl(trylast=True) def pytest_configure(config: pytest.Config) -> None: - if _is_xdist_controller(config): - return - state = _capture_state(config) - session_capture = state.capture if state is not None else None - plugin = s3_output.register_plugin(config, session_capture=session_capture) - if plugin is not None and state is not None: - state.claimed = True + if not _is_xdist_controller(config): + s3_output.register_plugin(config) diff --git a/tests/unittest/pytest.ini b/tests/unittest/pytest.ini index 5eb84b512ab5..8498d7e3f153 100644 --- a/tests/unittest/pytest.ini +++ b/tests/unittest/pytest.ini @@ -1,6 +1,8 @@ [pytest] xdist_start_method = spawn asyncio_default_fixture_loop_scope = module +log_format = %(asctime)s.%(msecs)03d %(levelname)-8s %(name)s:%(filename)s:%(lineno)d %(message)s +log_date_format = %Y-%m-%d %H:%M:%S threadleak = True # ThreadPoolExecutor-\d+_\d+ excludes worker threads leaked by torch._dynamo/torch.compile. # Thread-\d+ \(_manager_spawn\) is the MPIPoolExecutor manager thread of a pool cached diff --git a/tests/unittest/test_s3_output.py b/tests/unittest/test_s3_output.py index 899ad5a8fe27..60948006ad9e 100644 --- a/tests/unittest/test_s3_output.py +++ b/tests/unittest/test_s3_output.py @@ -13,461 +13,534 @@ # See the License for the specific language governing permissions and # limitations under the License. -import io -import logging -import os -import subprocess -import sys +import json +import threading +from pathlib import Path from types import SimpleNamespace import pytest -from test_common import s3_output_hooks -from test_common.s3_output import ( - FDRedirector, - FileSlice, - FileSliceReader, - SessionCapture, - UploadLogPlugin, -) +from test_common import s3_output, s3_output_hooks +from test_common.s3_output import UploadLogPlugin class Report: - def __init__(self, when=None, outcome=None): - self.sections = [] + def __init__( + self, + sections, + nodeid="test_module.py::test_case", + when="call", + outcome="passed", + rerun=None, + ): + self.sections = sections + self.nodeid = nodeid self.when = when self.outcome = outcome + if rerun is not None: + self.rerun = rerun -class CaptureContext: +class RecordingS3Client: def __init__(self): - self.closed = False + self.uploads = [] + + def upload_file(self, filepath, bucket, object_key, ExtraArgs=None): + content = Path(filepath).read_bytes() + self.uploads.append((content, bucket, object_key, ExtraArgs)) - def __exit__(self, exc_type, exc_val, exc_tb): - self.closed = True +class FailingS3Client: + def upload_file(self, filepath, bucket, object_key, ExtraArgs=None): + raise RuntimeError("upload error") -class RecordingS3Client: + +class BlockingS3Client(RecordingS3Client): def __init__(self): - self.uploads = [] - self.fileobj_uploads = [] + super().__init__() + self.started = threading.Event() + self.release = threading.Event() def upload_file(self, filepath, bucket, object_key, ExtraArgs=None): - self.uploads.append((filepath, bucket, object_key, ExtraArgs)) + self.started.set() + assert self.release.wait(timeout=10) + super().upload_file(filepath, bucket, object_key, ExtraArgs) + - def upload_fileobj(self, fileobj, bucket, object_key, ExtraArgs=None): - self.fileobj_uploads.append((fileobj.read(), bucket, object_key, ExtraArgs)) +class ConcurrentS3Client(RecordingS3Client): + def __init__(self, workers): + super().__init__() + self.barrier = threading.Barrier(workers) + + def upload_file(self, filepath, bucket, object_key, ExtraArgs=None): + self.barrier.wait(timeout=5) + super().upload_file(filepath, bucket, object_key, ExtraArgs) + + +class PluginManager: + def __init__(self): + self.plugins = {} + + def getplugin(self, name): + return self.plugins.get(name) + + def register(self, plugin, name): + self.plugins[name] = plugin + + +class Config: + def __init__(self, **options): + self.options = options + self.option = SimpleNamespace(numprocesses=options.get("numprocesses")) + self.known_args_namespace = self.option + self.pluginmanager = PluginManager() + + def getoption(self, name, default=None): + return self.options.get(name, default) def make_plugin( tmp_path, - inline_output_max_bytes, - capture_mode="timestamped", - session_capture=None, + inline_output_max_bytes=256, + skip_upload=True, + upload_mode="sync", + upload_workers=8, ): return UploadLogPlugin( endpoint_url="https://example.com", aws_access_key_id="user", - aws_secret_access_key=None, + aws_secret_access_key=None if skip_upload else "secret", bucket="bucket", upload_path="logs", output_path=str(tmp_path), - skip_upload=True, - capture_mode=capture_mode, + skip_upload=skip_upload, + upload_mode=upload_mode, + upload_workers=upload_workers, inline_output_max_bytes=inline_output_max_bytes, - session_capture=session_capture, ) -def write_log(tmp_path, test_name, filename, content): - test_dir = tmp_path / test_name - test_dir.mkdir() - (test_dir / filename).write_text(content, encoding="utf-8") +def make_uploading_plugin(tmp_path, monkeypatch, client, **kwargs): + monkeypatch.setattr(s3_output, "_create_s3_client", lambda *args: client) + return make_plugin(tmp_path, skip_upload=False, **kwargs) -def complete_makereport_hook(plugin, nodeid, report): - item = type("Item", (), {"nodeid": nodeid})() - hook = plugin.pytest_runtest_makereport(item, call=None) +def process_report(plugin, report): + hook = plugin.pytest_runtest_logreport(report) next(hook) - with pytest.raises(StopIteration) as stop: - hook.send(report) - assert stop.value.value is report - + with pytest.raises(StopIteration): + next(hook) -def test_capture_stays_open_after_successful_setup_report(tmp_path): - nodeid = "test_module.py::test_case" - plugin = make_plugin(tmp_path, inline_output_max_bytes=256) - capture = CaptureContext() - plugin._active_capture[nodeid] = {"stdout_redir": capture} - - complete_makereport_hook(plugin, nodeid, Report(when="setup", outcome="passed")) - - assert not capture.closed - assert nodeid in plugin._active_capture - plugin._close_capture(plugin._active_capture.pop(nodeid)) - - -@pytest.mark.parametrize( - ("when", "outcome"), - [ - ("call", "passed"), - ("call", "failed"), - ("setup", "failed"), - ("setup", "skipped"), - ], -) -def test_capture_closes_before_terminal_report(tmp_path, when, outcome): - nodeid = "test_module.py::test_case" - plugin = make_plugin(tmp_path, inline_output_max_bytes=256) - capture = CaptureContext() - plugin._active_capture[nodeid] = {"stdout_redir": capture} - complete_makereport_hook(plugin, nodeid, Report(when=when, outcome=outcome)) +def test_small_stdout_remains_inline(tmp_path): + plugin = make_plugin(tmp_path, inline_output_max_bytes=4) + report = Report([("Captured stdout call", "ok\n")]) - assert capture.closed - assert nodeid not in plugin._active_capture + process_report(plugin, report) + plugin.pytest_runtest_logfinish(report.nodeid, None) + assert report.sections == [("Captured stdout call", "ok\n")] + assert not s3_output._spool_root(str(tmp_path)).exists() -def test_teardown_fallback_warns_when_capture_is_still_active(tmp_path, caplog): - nodeid = "test_module.py::test_case" - plugin = make_plugin(tmp_path, inline_output_max_bytes=256) - capture = CaptureContext() - plugin._active_capture[nodeid] = {"stdout_redir": capture} - item = type("Item", (), {"nodeid": nodeid})() - caplog.set_level(logging.WARNING, logger="test_common.s3_output") - hook = plugin.pytest_runtest_teardown(item, nextitem=None) - next(hook) +def test_stdout_at_threshold_is_replaced_with_url(tmp_path): + plugin = make_plugin(tmp_path, inline_output_max_bytes=4) + report = Report([("Captured stdout call", "four")]) - assert capture.closed - assert nodeid not in plugin._active_capture - assert "remained active until teardown" in caplog.text + process_report(plugin, report) - with pytest.raises(StopIteration): - next(hook) + section_name, section_content = report.sections[0] + assert section_name == "Captured stdout" + assert "4 bytes (upload skipped" in section_content + assert "/stdout.log" in section_content + assert "four" not in section_content -def test_session_capture_closes_after_teardown(tmp_path): - nodeid = "test_module.py::test_case" - plugin = make_plugin(tmp_path, inline_output_max_bytes=256, capture_mode="session") - capture = CaptureContext() - plugin._active_capture[nodeid] = {"stdout_redir": capture} - item = type("Item", (), {"nodeid": nodeid})() +def test_inline_threshold_applies_to_combined_stream(tmp_path): + plugin = make_plugin(tmp_path, inline_output_max_bytes=8) + setup_section = ("Captured stdout setup", "abc") + call_section = ("Captured stdout call", "defgh") + setup_report = Report([setup_section], when="setup") + call_report = Report([setup_section, call_section]) + teardown_report = Report([setup_section, call_section], when="teardown") - hook = plugin.pytest_runtest_teardown(item, nextitem=None) - next(hook) - assert not capture.closed + process_report(plugin, setup_report) + process_report(plugin, call_report) + process_report(plugin, teardown_report) - with pytest.raises(StopIteration): - next(hook) - assert capture.closed - assert nodeid not in plugin._active_capture + assert setup_report.sections == [setup_section] + assert len(call_report.sections) == 1 + assert "/stdout.log" in call_report.sections[0][1] + assert len(teardown_report.sections) == 1 + stdout_file = next(s3_output._spool_root(str(tmp_path)).rglob("stdout.log")) + assert stdout_file.read_text(encoding="utf-8") == "abcdefgh" -def test_small_stdout_is_inlined_without_upload(tmp_path): - test_name = "test-small" - write_log(tmp_path, test_name, "stdout.log", "ok\n") - plugin = make_plugin(tmp_path, inline_output_max_bytes=4) - report = Report() +def test_logging_section_is_uploaded_even_when_small(tmp_path): + plugin = make_plugin(tmp_path, inline_output_max_bytes=256) + report = Report([("Captured log call", "log\n")]) - plugin.upload_and_report(report, test_name, "stdout.log", "Captured stdout") + process_report(plugin, report) - assert report.sections == [("Captured stdout", "ok\n")] - assert plugin._deferred_uploads == [] - assert not (tmp_path / test_name).exists() + assert "upload skipped" in report.sections[0][1] + assert "/logging.log" in report.sections[0][1] -def test_empty_stdout_is_removed_after_report(tmp_path): - test_name = "test-empty" - write_log(tmp_path, test_name, "stdout.log", "") - plugin = make_plugin(tmp_path, inline_output_max_bytes=4) - report = Report() +def test_sync_upload_transforms_native_sections(tmp_path, monkeypatch): + client = RecordingS3Client() + plugin = make_uploading_plugin( + tmp_path, + monkeypatch, + client, + inline_output_max_bytes=0, + ) + report = Report( + [ + ("Captured stdout setup", "setup output\n"), + ("custom", "keep me"), + ("Captured stderr call", "call error\n"), + ], + when="teardown", + ) - plugin.upload_and_report(report, test_name, "stdout.log", "Captured stdout") + process_report(plugin, report) + plugin.pytest_sessionfinish(None, 0) - assert report.sections == [("Captured stdout", "")] - assert plugin._deferred_uploads == [] - assert not (tmp_path / test_name).exists() + assert [upload[0] for upload in client.uploads] == [ + b"setup output\n", + b"call error\n", + ] + assert client.uploads[0][2].endswith("/stdout.log") + assert client.uploads[1][2].endswith("/stderr.log") + assert report.sections[1] == ("custom", "keep me") + assert all("uploaded to" in report.sections[index][1] for index in (0, 2)) + assert not s3_output._spool_root(str(tmp_path)).exists() -def test_large_stdout_keeps_existing_upload_report(tmp_path): - test_name = "test-large" - write_log(tmp_path, test_name, "stdout.log", "large output") - plugin = make_plugin(tmp_path, inline_output_max_bytes=3) - report = Report() +def test_duplicate_capture_sections_share_one_object(tmp_path): + plugin = make_plugin(tmp_path, inline_output_max_bytes=0) + report = Report( + [ + ("Captured stdout call", "first"), + ("Captured stdout call", "second"), + ] + ) - plugin.upload_and_report(report, test_name, "stdout.log", "Captured stdout") + process_report(plugin, report) assert len(report.sections) == 1 - section_name, section_content = report.sections[0] - assert section_name == "Captured stdout" - assert "upload skipped" in section_content - assert "large output" not in section_content - assert (tmp_path / test_name / "stdout.log").exists() - - -def test_uploaded_stdout_is_removed_after_sync_upload(tmp_path): - test_name = "test-uploaded" - write_log(tmp_path, test_name, "stdout.log", "large output") - plugin = make_plugin(tmp_path, inline_output_max_bytes=3) - plugin.skip_upload = False - plugin.s3 = RecordingS3Client() - report = Report() - - plugin.upload_and_report(report, test_name, "stdout.log", "Captured stdout") - - assert len(plugin.s3.uploads) == 1 - assert plugin.s3.uploads[0][1:] == ( - "bucket", - "logs/test-uploaded/stdout.log", - {"ContentType": "text/plain"}, + assert "/stdout.log" in report.sections[0][1] + stdout_file = next(s3_output._spool_root(str(tmp_path)).rglob("stdout.log")) + assert stdout_file.read_text(encoding="utf-8") == "firstsecond" + + +def test_cumulative_capture_sections_are_uploaded_once(tmp_path, monkeypatch): + client = RecordingS3Client() + plugin = make_uploading_plugin( + tmp_path, + monkeypatch, + client, + inline_output_max_bytes=0, ) - assert len(report.sections) == 1 - assert "uploaded to" in report.sections[0][1] - assert not (tmp_path / test_name).exists() - - -def test_deferred_stdout_is_removed_after_upload_finishes(tmp_path): - test_name = "test-deferred" - write_log(tmp_path, test_name, "stdout.log", "large output") - plugin = make_plugin(tmp_path, inline_output_max_bytes=3) - plugin.skip_upload = False - plugin.upload_mode = "deferred" - plugin.s3 = RecordingS3Client() - report = Report() - - plugin.upload_and_report(report, test_name, "stdout.log", "Captured stdout") - - assert (tmp_path / test_name / "stdout.log").exists() - assert len(plugin._deferred_uploads) == 1 - - plugin.pytest_sessionfinish(session=None, exitstatus=0) - - assert len(plugin.s3.uploads) == 1 - assert plugin._deferred_uploads == [] - assert not (tmp_path / test_name).exists() - - -def test_file_slice_upload_reads_only_test_range(tmp_path): - test_name = "test-slice" - spool = tmp_path / "stdout-spool.log" - prefix = b"previous test\n" - content = b"parent\nchild\nparent again\n" - spool.write_bytes(prefix + content + b"next test\n") - plugin = make_plugin(tmp_path, inline_output_max_bytes=3, capture_mode="session") - plugin.skip_upload = False - plugin.s3 = RecordingS3Client() - plugin._captured_slices[(test_name, "stdout.log")] = FileSlice( - str(spool), len(prefix), len(content) + setup_section = ("Captured stdout setup", "setup output\n") + call_section = ("Captured stdout call", "call output\n") + teardown_section = ("Captured stderr teardown", "teardown error\n") + + setup_report = Report([setup_section], when="setup") + call_report = Report( + [setup_section, call_section], + when="call", + outcome="failed", + ) + teardown_report = Report( + [setup_section, call_section, teardown_section], + when="teardown", ) - report = Report() - plugin.upload_and_report(report, test_name, "stdout.log", "Captured stdout") + process_report(plugin, setup_report) + process_report(plugin, call_report) + assert client.uploads == [] + process_report(plugin, teardown_report) + plugin.pytest_runtest_logfinish(setup_report.nodeid, None) + plugin.pytest_sessionfinish(None, 0) - assert plugin.s3.fileobj_uploads == [ - ( - content, - "bucket", - "logs/test-slice/stdout.log", - {"ContentType": "text/plain"}, - ) + assert [upload[0] for upload in client.uploads] == [ + b"setup output\ncall output\n", + b"teardown error\n", ] - assert "uploaded to" in report.sections[0][1] - - -def test_file_slice_reader_supports_upload_size_probe(tmp_path): - spool = tmp_path / "stdout-spool.log" - spool.write_bytes(b"prefix\ntest output\nsuffix\n") - file_slice = FileSlice(str(spool), len(b"prefix\n"), len(b"test output\n")) - - with io.BufferedReader(FileSliceReader(file_slice)) as source_file: - assert source_file.tell() == 0 - assert source_file.seek(0, os.SEEK_END) == file_slice.size - assert source_file.tell() == file_slice.size - assert source_file.seek(0) == 0 - assert source_file.read() == b"test output\n" - - -def test_session_capture_preserves_parent_and_child_stdout_order(tmp_path): - capture = SessionCapture(str(tmp_path)) - capture.start() - child = None - try: - child = subprocess.Popen( - [ - sys.executable, - "-c", - "import os, sys; sys.stdin.buffer.read(1); " - "os.write(1, b'child stdout\\n'); os.write(2, b'child stderr\\n')", - ], - stdin=subprocess.PIPE, - ) - offsets = capture.snapshot() - os.write(1, b"parent before\n") - child.stdin.write(b"x") - child.stdin.close() - assert child.wait(timeout=10) == 0 - os.write(1, b"parent after\n") - slices = capture.slices_since(offsets) - finally: - if child is not None and child.poll() is None: - child.kill() - child.wait() - capture.stop() - - with io.BufferedReader(FileSliceReader(slices["stdout.log"])) as stdout_file: - assert stdout_file.read() == b"parent before\nchild stdout\nparent after\n" - with io.BufferedReader(FileSliceReader(slices["stderr.log"])) as stderr_file: - assert stderr_file.read() == b"child stderr\n" - capture.remove_files() - - -class _EarlyConfig: - def __init__(self, capture="no", numprocesses=None): - self.option = SimpleNamespace(numprocesses=numprocesses) - self.known_args_namespace = SimpleNamespace( - capture=capture, - numprocesses=numprocesses, - ) - self.cleanups = [] - - def add_cleanup(self, cleanup): - self.cleanups.append(cleanup) - - -def test_early_capture_preserves_conftest_child_output_for_case(tmp_path, monkeypatch): - monkeypatch.delenv("PYTEST_XDIST_WORKER", raising=False) - early_config = _EarlyConfig() - initial_hook = s3_output_hooks.pytest_load_initial_conftests( - early_config, - [ - "-s", - "--s3-upload-path=logs", - f"--output-dir={tmp_path}", - ], - ) - child = None - capture = None - try: - next(initial_hook) - state = s3_output_hooks._capture_state(early_config) - assert state is not None - capture = state.capture - assert capture is not None - child = subprocess.Popen( - [ - sys.executable, - "-c", - "import os, sys; sys.stdin.buffer.read(1); " - "os.write(1, b'child stdout\\n'); os.write(2, b'child stderr\\n')", - ], - stdin=subprocess.PIPE, - ) - - with pytest.raises(StopIteration): - next(initial_hook) - - plugin = make_plugin( - tmp_path, - inline_output_max_bytes=0, - capture_mode="session", - session_capture=capture, - ) - session = SimpleNamespace(config=SimpleNamespace()) - session_hook = plugin.pytest_sessionstart(session) - next(session_hook) - with pytest.raises(StopIteration): - next(session_hook) - - offsets = capture.snapshot() - os.write(1, b"parent before\n") - child.stdin.write(b"x") - child.stdin.close() - assert child.wait(timeout=10) == 0 - os.write(1, b"parent after\n") - slices = capture.slices_since(offsets) - capture.stop() - - with io.BufferedReader(FileSliceReader(slices["stdout.log"])) as stdout_file: - assert stdout_file.read() == b"parent before\nchild stdout\nparent after\n" - with io.BufferedReader(FileSliceReader(slices["stderr.log"])) as stderr_file: - assert stderr_file.read() == b"child stderr\n" - finally: - if child is not None and child.poll() is None: - child.kill() - child.wait() - if capture is not None: - capture.stop() - for cleanup in early_config.cleanups: - cleanup() - - -def test_early_capture_skips_xdist_controller(tmp_path, monkeypatch): - monkeypatch.delenv("PYTEST_XDIST_WORKER", raising=False) - early_config = _EarlyConfig(numprocesses=2) - initial_hook = s3_output_hooks.pytest_load_initial_conftests( - early_config, - [ - "-s", - "-n", - "2", - "--s3-upload-path=logs", - f"--output-dir={tmp_path}", - ], + assert [upload[2].rsplit("/", 1)[-1] for upload in client.uploads] == [ + "stdout.log", + "stderr.log", + ] + assert "Last 200 lines:" in call_report.sections[0][1] + assert "setup output" in call_report.sections[0][1] + assert "call output" in call_report.sections[0][1] + assert "Last 200 lines:" not in teardown_report.sections[0][1] + assert sum("/stdout.log" in content for _, content in teardown_report.sections) == 1 + assert sum("/stderr.log" in content for _, content in teardown_report.sections) == 1 + assert len(teardown_report.sections) == 2 + assert all(content.endswith("\n") for _, content in teardown_report.sections) + + +def test_same_nodeid_rerun_gets_distinct_test_path(tmp_path, monkeypatch): + monkeypatch.setattr(s3_output.time, "time", lambda: 1234) + plugin = make_plugin(tmp_path, inline_output_max_bytes=0) + nodeid = "test_module.py::test_case" + + plugin.pytest_runtest_logstart(nodeid, None) + first_name = plugin._test_names[nodeid] + plugin.pytest_runtest_logfinish(nodeid, None) + plugin.pytest_runtest_logstart(nodeid, None) + second_name = plugin._test_names[nodeid] + + assert second_name == f"{first_name}-1" + + +def test_failed_report_keeps_only_recent_bounded_output(tmp_path): + plugin = make_plugin(tmp_path, inline_output_max_bytes=0) + content = "".join(f"line-{index:03d}\n" for index in range(250)) + report = Report( + [("Captured stdout call", content)], + outcome="failed", ) - next(initial_hook) - assert s3_output_hooks._capture_state(early_config) is None - with pytest.raises(StopIteration): - next(initial_hook) + process_report(plugin, report) + section_content = report.sections[0][1] + assert "Last 200 lines:" in section_content + assert "line-000" not in section_content + assert "line-249" in section_content -def test_s3_plugin_registration_runs_on_xdist_worker_only(monkeypatch): - monkeypatch.delenv("PYTEST_XDIST_WORKER", raising=False) - registered_configs = [] + large_line = "x" * 70000 + report = Report( + [("Captured stderr call", large_line)], + nodeid="test_module.py::test_other", + outcome="failed", + ) + process_report(plugin, report) + assert "... [truncated]" in report.sections[0][1] + assert len(report.sections[0][1].encode()) < 66000 + + +def test_deferred_upload_starts_before_session_finish(tmp_path, monkeypatch): + client = BlockingS3Client() + plugin = make_uploading_plugin( + tmp_path, + monkeypatch, + client, + inline_output_max_bytes=0, + upload_mode="deferred", + upload_workers=1, + ) + report = Report([("Captured stdout call", "background output")]) + + process_report(plugin, report) + plugin.pytest_runtest_logfinish(report.nodeid, None) + + assert client.started.wait(timeout=5) + assert "scheduled for upload" in report.sections[0][1] + client.release.set() + plugin.pytest_sessionfinish(None, 0) + assert client.uploads[0][0] == b"background output" + assert not s3_output._spool_root(str(tmp_path)).exists() + + +def test_deferred_upload_reuses_cumulative_section(tmp_path, monkeypatch): + client = BlockingS3Client() + plugin = make_uploading_plugin( + tmp_path, + monkeypatch, + client, + inline_output_max_bytes=0, + upload_mode="deferred", + upload_workers=1, + ) + section = ("Captured stdout call", "background output") + call_report = Report([section], outcome="failed") + teardown_report = Report([section], when="teardown") + + process_report(plugin, call_report) + process_report(plugin, teardown_report) + + assert client.started.wait(timeout=5) + assert len(plugin._pending_uploads) == 1 + assert ( + call_report.sections[0][1].splitlines()[0] + == (teardown_report.sections[0][1].splitlines()[0]) + ) + client.release.set() + plugin.pytest_sessionfinish(None, 0) + assert [upload[0] for upload in client.uploads] == [b"background output"] + + +def test_rerun_keeps_one_url_per_attempt(tmp_path, monkeypatch): + monkeypatch.setattr(s3_output.time, "time", lambda: 1234) + client = RecordingS3Client() + plugin = make_uploading_plugin( + tmp_path, + monkeypatch, + client, + inline_output_max_bytes=0, + ) + nodeid = "test_module.py::test_case" + first_sections = [ + ("Captured stdout call", "first stdout\n"), + ("Captured stderr call", "first stderr\n"), + ] + all_sections = first_sections + [ + ("Captured stdout call", "second stdout\n"), + ("Captured stderr call", "second stderr\n"), + ] - def register_plugin(config, session_capture=None): - registered_configs.append(config) - return object() + plugin.pytest_runtest_logstart(nodeid, None) + first_report = Report( + first_sections, + nodeid=nodeid, + outcome="rerun", + rerun=0, + ) + process_report(plugin, first_report) + plugin.pytest_runtest_logfinish(nodeid, None) + + plugin.pytest_runtest_logstart(nodeid, None) + second_report = Report( + list(all_sections), + nodeid=nodeid, + outcome="failed", + rerun=1, + ) + process_report(plugin, second_report) + teardown_report = Report( + list(all_sections), + nodeid=nodeid, + when="teardown", + rerun=1, + ) + process_report(plugin, teardown_report) + plugin.pytest_runtest_logfinish(nodeid, None) + plugin.pytest_sessionfinish(None, 0) + + assert [upload[0] for upload in client.uploads] == [ + b"first stdout\n", + b"first stderr\n", + b"second stdout\n", + b"second stderr\n", + ] + object_keys = [upload[2] for upload in client.uploads] + assert object_keys[0].endswith("/stdout.log") + assert object_keys[1].endswith("/stderr.log") + assert object_keys[2].endswith("/stdout-attempt-2.log") + assert object_keys[3].endswith("/stderr-attempt-2.log") + assert object_keys[0].rsplit("/", 1)[0] != object_keys[2].rsplit("/", 1)[0] + assert all(content.endswith("\n") for _, content in teardown_report.sections) + assert "Last 200 lines:" not in teardown_report.sections[0][1] + + +def test_parent_drain_retries_upload_left_by_failed_process(tmp_path, monkeypatch): + plugin = make_uploading_plugin( + tmp_path, + monkeypatch, + FailingS3Client(), + inline_output_max_bytes=0, + ) + report = Report([("Captured stdout call", "recover me")], when="teardown") + process_report(plugin, report) + assert "upload failed" in report.sections[0][1] + + config_path = Path(plugin._spool_config_path) + config = json.loads(config_path.read_text(encoding="utf-8")) + config["pid"] = 99999999 + config_path.write_text(json.dumps(config), encoding="utf-8") + + client = RecordingS3Client() + monkeypatch.setattr(s3_output, "_create_s3_client", lambda *args: client) + assert s3_output.drain_pending_uploads(str(tmp_path), secret_key="secret") + assert client.uploads[0][0] == b"recover me" + assert client.uploads[0][2].endswith("/stdout.log") + assert not s3_output._spool_root(str(tmp_path)).exists() + + +def test_parent_drain_uses_configured_upload_workers(tmp_path, monkeypatch): + plugin = make_uploading_plugin( + tmp_path, + monkeypatch, + RecordingS3Client(), + upload_workers=2, + ) + plugin._append_spool_file("test", "stdout-call.log", "stdout") + plugin._append_spool_file("test", "stderr-call.log", "stderr") + config_path = Path(plugin._spool_config_path) + config = json.loads(config_path.read_text(encoding="utf-8")) + config["pid"] = 99999999 + config_path.write_text(json.dumps(config), encoding="utf-8") + + client = ConcurrentS3Client(workers=2) + monkeypatch.setattr(s3_output, "_create_s3_client", lambda *args: client) + + assert s3_output.drain_pending_uploads(str(tmp_path), secret_key="secret") + assert sorted(upload[0] for upload in client.uploads) == [b"stderr", b"stdout"] + assert not s3_output._spool_root(str(tmp_path)).exists() + + +def test_parent_drain_skips_live_pytest_process(tmp_path, monkeypatch): + client = RecordingS3Client() + plugin = make_uploading_plugin( + tmp_path, + monkeypatch, + client, + inline_output_max_bytes=0, + upload_mode="deferred", + ) + plugin._append_spool_file("test", "stdout-call.log", "still active") + + assert s3_output.drain_pending_uploads(str(tmp_path), secret_key="secret") + assert client.uploads == [] + assert Path(plugin._spool_config_path).exists() + plugin.pytest_sessionfinish(None, 0) + + +def test_register_plugin_requires_native_fd_capture(tmp_path): + config = Config( + **{ + "--s3-upload-path": "logs", + "--output-dir": str(tmp_path), + "--s3-skip-upload": True, + "capture": "no", + } + ) - monkeypatch.setattr(s3_output_hooks.s3_output, "register_plugin", register_plugin) - controller_config = _EarlyConfig(numprocesses=2) - s3_output_hooks.pytest_configure(controller_config) - assert registered_configs == [] + with pytest.raises(ValueError, match="requires pytest --capture=fd"): + s3_output.register_plugin(config) + + +def test_register_plugin_uses_report_transformer(tmp_path): + config = Config( + **{ + "--s3-upload-path": "logs", + "--output-dir": str(tmp_path), + "--s3-skip-upload": True, + "--s3-endpoint": "https://example.com", + "--s3-username": "user", + "--s3-bucket": "bucket", + "capture": "fd", + } + ) - monkeypatch.setenv("PYTEST_XDIST_WORKER", "gw0") - worker_config = _EarlyConfig(numprocesses=2) - s3_output_hooks.pytest_configure(worker_config) - assert registered_configs == [worker_config] + plugin = s3_output.register_plugin(config) + assert isinstance(plugin, UploadLogPlugin) + assert config.pluginmanager.getplugin("upload_log_plugin") is plugin -def test_small_log_file_is_not_inlined(tmp_path): - test_name = "test-log" - write_log(tmp_path, test_name, "logging.log", "ok\n") - plugin = make_plugin(tmp_path, inline_output_max_bytes=256) - report = Report() - plugin.upload_and_report(report, test_name, "logging.log", "Captured log") +def test_s3_hook_skips_xdist_controller(monkeypatch): + registered = [] + monkeypatch.setattr(s3_output_hooks.s3_output, "register_plugin", registered.append) + monkeypatch.delenv("PYTEST_XDIST_WORKER", raising=False) - assert len(report.sections) == 1 - section_name, section_content = report.sections[0] - assert section_name == "Captured log" - assert "upload skipped" in section_content - - -def test_fd_redirector_reader_thread_is_daemon_when_pipe_writer_lingers(tmp_path): - redir = FDRedirector(1, str(tmp_path / "stdout.log")) - proc = None - try: - redir.__enter__() - proc = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(30)"]) - redir.__exit__(None, None, None) - - assert redir._reader_thread is not None - assert redir._reader_thread.is_alive() - assert redir._reader_thread.daemon - finally: - if proc is not None: - proc.kill() - proc.wait() - if redir._reader_thread is not None: - redir._reader_thread.join(timeout=2.0) + controller = Config(numprocesses=2) + s3_output_hooks.pytest_configure(controller) + assert registered == [] + + monkeypatch.setenv("PYTEST_XDIST_WORKER", "gw0") + worker = Config(numprocesses=2) + s3_output_hooks.pytest_configure(worker) + assert registered == [worker] diff --git a/tests/unittest/tools/test_test_to_stage_mapping.py b/tests/unittest/tools/test_test_to_stage_mapping.py index 0ca9bdfcd94a..18c982bab4ec 100644 --- a/tests/unittest/tools/test_test_to_stage_mapping.py +++ b/tests/unittest/tools/test_test_to_stage_mapping.py @@ -82,36 +82,6 @@ def test_data_availability(stage_query): print(f"Max samples configured: {MAX_SAMPLES}") -def test_s3_stdout_echo_requires_explicit_opt_in(): - """Keep Jenkins pytest progress readable unless live log echo is requested.""" - with open(GROOVY, 'r') as f: - lines = f.readlines() - - assert any( - 'ENABLE_S3_ECHO_STDOUT = params.enableS3EchoStdout != null ? params.enableS3EchoStdout : false' - in line for line in lines) - - echo_lines = [ - idx for idx, line in enumerate(lines) if '"--s3-echo-stdout"' in line - ] - assert echo_lines, 'Expected at least one opt-in --s3-echo-stdout path' - - for idx in echo_lines: - context = lines[max(0, idx - 3):idx] - assert any('if (ENABLE_S3_ECHO_STDOUT)' in line for line in context) - - progress_lines = [ - idx for idx, line in enumerate(lines) - if 'console_output_style=progress-even-when-capture-no' in line - ] - assert progress_lines, 'Expected upload-only pytest progress configuration' - - for idx in progress_lines: - context = lines[max(0, idx - 3):idx] - assert any('if (ENABLE_UPLOAD_TEST_RESULTS)' in line - for line in context) - - def test_documented_stage_examples_are_live(stage_query): """Documented --stages examples must name stages that still exist in CI.""" sources = [