diff --git a/Jenkinsfile b/Jenkinsfile index cfa4963f841..29d9ce93e39 100644 --- a/Jenkinsfile +++ b/Jenkinsfile @@ -87,6 +87,24 @@ void updateRunStage() { println("updateRunStage: Build cause: ${buildCauses}") println("updateRunStage: Started by user: ${startedByUser()}") + // Force an NLT-only run profile for targeted reliability testing. + if (params.CI_NLT_ONLY) { + println('updateRunStage: Detected CI_NLT_ONLY, enabling only NLT and required build stages') + List nltOnlyStages = [ + 'Pre-build', + 'Build', + 'Build on EL 9', + 'Unit Tests', + 'NLT' + ] + for (stage in runStage.keySet()) { + runStage[stage] = stage in nltOnlyStages + reasons[stage] = 'CI_NLT_ONLY' + } + displayRunStage(reasons) + return + } + // Handle landing builds if (startedByLanding()) { println('updateRunStage: Detected landing build, overwriting defaults') @@ -539,6 +557,9 @@ pipeline { booleanParam(name: 'CI_BUILD_PACKAGES_ONLY', defaultValue: false, description: 'Build RPM and DEB packages, Skip unit tests.') + booleanParam(name: 'CI_NLT_ONLY', + defaultValue: true, + description: 'Run only stages needed for NLT: Build on EL 9 and NLT path.') booleanParam(name: 'CI_ALLOW_UNSTABLE_TEST', defaultValue: false, description: 'Continue testing if a previous stage is Unstable') @@ -994,8 +1015,16 @@ pipeline { label params.CI_NLT_1_LABEL } steps { - // NLT memchecks the valgrind-tagged build, not the shared -race one. - unstash 'opt-daos-valgrind' + // Prefer the valgrind-tagged build when available, but allow + // NLT to run without it (e.g. CI_PR_REPOS-only workflows). + script { + try { + unstash 'opt-daos-valgrind' + echo 'Using opt-daos-valgrind stash for NLT' + } catch (hudson.AbortException err) { + echo "opt-daos-valgrind stash not available, falling back to opt-daos.tar (${err.message})" + } + } job_step_update( unitTest(timeout_time: 60 * cachedCommitPragma(pragma: 'NLT-repeat', def_val: '1').toInteger(), @@ -1028,7 +1057,7 @@ pipeline { } post { always { - unitTestPost artifacts: ['nlt_logs/'], + unitTestPost artifacts: ['nlt_logs/', 'vm_test/'], testResults: 'nlt-junit.xml', valgrind_stash: 'nlt-memcheck', valgrind_pattern: '*memcheck.xml', @@ -1194,7 +1223,7 @@ pipeline { ], unitTestPostArgs: [ /* groovylint-disable-next-line DuplicateListLiteral */ - artifacts: ['nlt_logs/'], + artifacts: ['nlt_logs/', 'vm_test/'], testResults: 'nlt-junit.xml', with_valgrind: '', FI: true], diff --git a/ci/unit/test_nlt_post.sh b/ci/unit/test_nlt_post.sh index db39d9c7c3d..b5f202c1e95 100755 --- a/ci/unit/test_nlt_post.sh +++ b/ci/unit/test_nlt_post.sh @@ -27,4 +27,5 @@ rsync -v -dpt -z -e "ssh $SSH_KEY_ARGS" jenkins@"$NODE":build/ \ --filter="include nlt-junit.xml" --filter="exclude *" ./ mkdir -p vm_test -mv nlt-errors.json vm_test/ +mv nlt-errors.json vm_test/ || echo "nlt-errors.json not found (NLT may have crashed)" +mv nlt-summary.json vm_test/ || echo "nlt-summary.json not found" diff --git a/docs/testing/nlt.md b/docs/testing/nlt.md new file mode 100644 index 00000000000..948e91bdfc1 --- /dev/null +++ b/docs/testing/nlt.md @@ -0,0 +1,252 @@ +# Node Local Test (NLT) + +`utils/node_local_test.py` is a single-node integration and smoke-test harness for DAOS. It boots +a local DAOS stack (server engine, agent, optional dfuse mount), runs a suite of functional and +fault-injection tests against it, analyzes all daemon and client logs for anomalies, and emits +structured CI artifacts. + +## Overview + +NLT runs as part of the CI pipeline on a dedicated VM. The entry point is +`ci/unit/test_nlt.sh`, which rsyncs the build to the node and runs +`ci/unit/test_nlt_node.sh` via SSH. The node script installs DAOS, creates a Python venv, +mounts `nlt_logs/` on tmpfs, and execs `node_local_test.py`. After the run, +`ci/unit/test_nlt_post.sh` rsyncs logs and result artifacts back to Jenkins. + +## Run modes + +`node_local_test.py` accepts one or more positional `mode` arguments: + +| Mode | What runs | +|------|-----------| +| `all` (default) | Full POSIX test suite + dfuse multi-mount + UNS overlay + pydaos KV tests + 3 dfuse FI tests | +| `fi` | Exhaustive allocation-failure sweep across many DAOS client commands (no POSIX suite) | +| `launch` | Start the server only and drop to a shell (for interactive debugging) | +| `set-fi` | Start the server, enable fault injection, and exit (used by other tooling) | + +The CI pipeline runs two separate stages using these modes: + +- **NLT stage**: `mode=all` with valgrind memcheck enabled. Runs the full functional suite plus + a small set of dfuse FI tests. Typical duration ~20 minutes. +- **Fault injection testing stage**: `mode=fi` with memcheck disabled, server logging at `WARN`, + and a 14-CPU VM. Runs the full allocation-failure sweep in parallel Docker containers. + Typical duration up to 4 hours. + +## Test suites + +### POSIX test suite (`mode=all`) + +`PosixTests` contains ~55 test methods that exercise dfuse and the DAOS POSIX layer. Each test +gets its own freshly created POSIX container for isolation. Tests run in parallel (up to 4 +threads) with slow tests (`test_uns_basic`, `test_daos_fs_tool`, `test_stable_cont_inode`) +sorted to the front to minimize wall time. + +Tests are decorated to control how they run: + +- `@needs_dfuse`: runs twice — once with caching disabled, once with caching enabled. +- `@needs_dfuse_with_opt`: similar but with configurable option variants. +- Plain methods: run once against the pool/container directly (no dfuse mount). + +This produces approximately 82 total test invocations from the ~55 base methods. + +After the POSIX suite, `mode=all` also runs: + +- `run_dfuse`: multi-mount dfuse stress tests. +- `run_duns_overlay_test`: UNS overlay tests. +- `test_pydaos_kv` / `test_pydaos_kv_obj_class`: Python pydaos KV API tests. +- `server.set_fi()`: a brief server-side fault injection pass. + +### Fault injection (FI) tests + +FI tests use the `AllocFailTest` class to verify that every `D_ALLOC()` call site in a given +code path handles allocation failure correctly. For each test: + +1. The command runs once baseline (no injection) to establish expected behaviour. +2. The command is re-run repeatedly, injecting a failure at allocation site `fid=2, 3, 4, ...` + in parallel batches. +3. The sweep stops when an iteration triggers no injection (`NLTestNoFi`), meaning all + allocation points have been exhausted. +4. Any run that crashes is automatically re-run under valgrind for leak and signal detail. + +**FI tests that run in `mode=fi`** (designed for parallel Docker execution): + +| Test | Command under test | +|------|--------------------| +| `test_dfuse_start` | dfuse startup error paths | +| `test_alloc_fail` | `daos cont list` | +| `test_fi_cont_query` | `daos cont query` | +| `test_fi_cont_check` | `daos cont check` | +| `test_fi_get_attr` | `daos cont get-attr` | +| `test_fi_list_attr` | `daos cont list-attr` | +| `test_fi_get_prop` | `daos cont get-prop` | +| `test_alloc_fail_copy` | `daos filesystem copy` (read + write) | +| `test_alloc_fail_copy_trunc` | `daos filesystem copy` with truncation | +| `test_alloc_cont_create` | `daos cont create` (with properties) | + +**FI tests that run in `mode=all`** (require a live dfuse mount, cannot run in Docker): + +| Test | Command under test | +|------|--------------------| +| `test_alloc_fail_cont_create` | `daos cont create --path` via UNS | +| `test_alloc_fail_cat` | `cat` via interception library (IL) | +| `test_alloc_fail_il_cp` | `cp` via interception library (IL) | + +### Restart and valgrind server checks + +After the main test pass (`mode=all`), NLT: + +1. Starts a second server instance (`test_class='restart'`) and immediately stops it — verifying + the server can cleanly restart against existing storage. +2. Optionally starts a third server instance under valgrind (`--server-valgrind`) and runs + basic pool and container queries to check for server-side leaks. + +## Log analysis (`cart_logtest.py`) + +After each command or daemon exits, `node_local_test.py` calls `log_test()`, which dynamically +imports `src/tests/ftest/cart/util/cart_logtest.py` and runs it against the DAOS debug log for +that process. `cart_logtest` parses the structured DAOS log format and flags anomalies. + +The `WarningsFactory` instance from `node_local_test.py` is injected into `cart_logtest` so +all findings are written directly into the appropriate warnings JSON file. + +### Severity levels + +| Severity | Meaning | +|----------|---------| +| `ERROR` | Process crashed, harness shutdown without clean close, teardown failure | +| `HIGH` | `ERR`-level log line in a strict-mode source file — likely a real DAOS bug | +| `NORMAL` | Anomaly that may indicate a problem: wrong error code, excessive logging, RPC lifecycle issue, or allocation failure logged twice within 5 lines of the same file during FI | +| `LOW` | Convention issue (e.g. error code formatted with `%d` instead of `DF_RC`) | + +### What `NORMAL: Logging allocation failure` means + +This specific entry is emitted when, during a fault injection run, the injected `D_ALLOC()` +failure causes a `-1009` (DER_NOMEM) log line to appear within 5 source lines of the injection +point in the same file. It indicates that the failure is being propagated and logged correctly — +the code is handling the OOM — but that two adjacent call sites both log the same error. This is +a log verbosity pattern, not a correctness bug. + +### Output file mapping + +| File | Written by | Content | +|------|-----------|---------| +| `nlt-errors.json` | `wf` (main harness + FI sweep) | General anomalies from all log analysis | +| `nlt-server-leaks.json` | `wf_server` | Server log analysis: leaks, opcode state, strict-mode warnings | +| `nlt-client-leaks.json` | `wf_client` | Client FI result checks (`mode=fi` only) | + +## CI quality gates + +Jenkins uses the Warnings Next Generation plugin (`recordIssues`) to evaluate `nlt-errors.json` +and `nlt-server-leaks.json` against a reference build from `master`. The gates are: + +| Gate | Threshold | Result | +|------|-----------|--------| +| Total `ERROR` severity | ≥ 1 | UNSTABLE | +| Total `HIGH` severity | ≥ 1 | UNSTABLE | +| **New** `NORMAL` severity (vs reference) | ≥ 1 | UNSTABLE | +| **New** `LOW` severity (vs reference) | ≥ 1 | UNSTABLE | +| Server leaks total (any severity) | ≥ 1 | UNSTABLE | + +"New" means an issue present in the current build that was not fingerprinted in the reference +master build. Jenkins fingerprints issues by file, line number, and message — so changing the +message text of an existing check will cause it to re-appear as "new" on the next build until +the reference catches up. + +Note that `NORMAL: Logging allocation failure` entries are expected in passing builds wherever +FI tests exercise code paths with adjacent error logging. These appear as "outstanding" (not +"new") against a stable reference. A PR that adds new `D_ALLOC()` call sites in such paths +will produce a new entry and trip the `NEW_NORMAL` gate. + +## Artifacts + +### Primary triage artifact + +- `nlt-summary.json`: compact run summary for humans and CI automation. + - Contains run metadata (`run_id`, mode, class name, repeat, engine count). + - Contains pass/fail result and top high-severity findings (`HIGH`/`ERROR`). + - Contains per-file issue counts for warnings JSON artifacts. + - Contains artifact-presence flags (`nlt-junit.xml`, `nlt_logs`, log-usage exports). + +### Compatibility artifacts + +- `nlt-junit.xml`: JUnit test-case output consumed by test-result publishers. +- `nlt-errors.json`: warnings stream for core log file errors. +- `nlt-server-leaks.json`: warnings stream focused on server leak checks. +- `nlt-client-leaks.json`: warnings stream focused on client leak checks (`mode=fi` only). + +All warnings JSON entries include a `runId` (UUID generated at startup) so findings can be +correlated across all outputs from the same NLT execution. + +### Log artifacts + +- `nlt_logs/dnt_*.log.bz2`: compressed DAOS debug log for every process invocation. Each file + is named with a `dnt_` prefix and a short descriptor of the test or daemon that produced it + (e.g. `dnt_dfuse_test_rename_caching_off_.log.bz2`, + `dnt_server_Server.first_0_.log.bz2`). +- `dnt*.memcheck.xml`: valgrind memcheck output per test invocation. +- `nltir.xml` / `nltr.json`: optional log-usage reports (generated when `--log-usage-save` / + `--log-usage-import` are passed). + +## Running NLT locally + +```bash +# Full functional suite (requires installed DAOS at /opt/daos) +python3 utils/node_local_test.py all + +# Run a single named test +python3 utils/node_local_test.py --test rename + +# Run without valgrind (faster) +python3 utils/node_local_test.py --memcheck no all + +# Run the FI sweep only +python3 utils/node_local_test.py fi + +# Start server interactively for debugging +python3 utils/node_local_test.py launch + +# List all available tests +python3 utils/node_local_test.py --test list +``` + +Key options: + +| Option | Default | Description | +|--------|---------|-------------| +| `--memcheck` | `some` | Valgrind coverage: `yes` (all), `some` (most), `no` (none) | +| `--server-debug` | `DEBUG` | Server log level | +| `--engine-count` | `1` | Number of DAOS engines to start | +| `--repeat` | `1` | Number of times to repeat the full test pass | +| `--test` | all | Run only specific named test(s) | +| `--exclude-test` | none | Exclude specific test(s) from the suite | +| `--dfuse-dir` | `/tmp` | Parent directory for dfuse mounts | + +## Primary triage artifact + +- `nlt-summary.json`: compact run summary for humans and CI automation. + - Contains run metadata (`run_id`, mode, class name, repeat, engine count). + - Contains pass/fail result and top high-severity findings (`HIGH`/`ERROR`). + - Contains per-file issue counts for warnings JSON artifacts. + - Contains artifact-presence flags (`nlt-junit.xml`, `nlt_logs`, log-usage exports). + +## Compatibility artifacts + +- `nlt-junit.xml`: JUnit test-case output consumed by test-result publishers. +- `nlt-errors.json`: warnings stream for core log file errors. +- `nlt-server-leaks.json`: warnings stream focused on server leak checks. +- `nlt-client-leaks.json`: warnings stream focused on client leak checks. + +Warnings JSON entries now include `runId` so findings can be correlated across all outputs from +the same NLT execution. + +## Log artifacts + +- `nlt_logs/*`: daemon/client/fault-injection logs and compressed `.bz2` variants. +- `dnt*.xml`: valgrind outputs where applicable. +- `nltir.xml` / `nltr.json`: optional log-usage reports when requested. + +## CI consumption guidance + +- Use `nlt-summary.json` as the first-stop triage surface. +- Keep consuming `nlt-junit.xml` for pass/fail test publishing. +- Keep warnings JSON consumption for backward compatibility and gating. diff --git a/mkdocs.yml b/mkdocs.yml index 95fe52b04e3..a2e9e2caae5 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -91,6 +91,7 @@ nav: - 'Environment Variables': 'admin/env_variables.md' - Test and Benchmarking: - 'Run DAOS Autotest': 'testing/autotest.md' + - 'Node Local Test': 'testing/nlt.md' - 'Run IOR and mdtest': 'testing/ior.md' - 'Run dbench': 'testing/dbench.md' - 'Run Datamover test': 'testing/datamover.md' diff --git a/src/tests/ftest/cart/util/cart_logtest.py b/src/tests/ftest/cart/util/cart_logtest.py old mode 100755 new mode 100644 index a500cf2046c..80340c351a6 --- a/src/tests/ftest/cart/util/cart_logtest.py +++ b/src/tests/ftest/cart/util/cart_logtest.py @@ -407,7 +407,7 @@ def _check_pid_from_log_file(self, pid, abort_on_warning, leak_wf, show_memleaks src_offset -= self.fi_location.lineno if 0 < src_offset < 5: show_line(line, 'NORMAL', - 'Logging allocation failure') + 'Logging allocation failure (same allocation failure logged within 5 lines)') if not line.get_msg().endswith("DER_NOMEM(-1009): 'Out of memory'"): show_line(line, 'LOW', 'Error does not use DF_RC') diff --git a/utils/node_local_test.py b/utils/node_local_test.py index 7d2e3191ab2..fd704937180 100755 --- a/utils/node_local_test.py +++ b/utils/node_local_test.py @@ -32,6 +32,7 @@ import resource import shutil import signal +import socket import stat import subprocess # nosec import sys @@ -99,6 +100,43 @@ def umount(path, background=False): return ret.returncode +def terminate_process(proc, + name, + graceful_signal=signal.SIGTERM, + graceful_timeout=5, + kill_timeout=5): + """Terminate a subprocess and reap it deterministically. + + Returns: + tuple[int|None, bool]: (returncode if reaped, had_to_escalate) + """ + if proc is None: + return (None, False) + + had_to_escalate = False + try: + proc.send_signal(graceful_signal) + except ProcessLookupError: + return (proc.poll(), False) + + try: + return (proc.wait(timeout=graceful_timeout), False) + except subprocess.TimeoutExpired: + had_to_escalate = True + print(f'Timeout stopping {name} with {graceful_signal.name}, sending SIGKILL') + + try: + proc.send_signal(signal.SIGKILL) + except ProcessLookupError: + return (proc.poll(), had_to_escalate) + + try: + return (proc.wait(timeout=kill_timeout), had_to_escalate) + except subprocess.TimeoutExpired: + print(f'Unable to reap {name} after SIGKILL') + return (None, had_to_escalate) + + class NLTConf(): """Helper class for configuration""" @@ -233,6 +271,7 @@ def __init__(self, self._class_id = class_id self.pending = [] self._running = True + self._lock = threading.RLock() # Save the filename of the object, as __file__ does not # work in __del__ self._file = __file__.lstrip('./') @@ -290,18 +329,20 @@ class and other metadata will be set automatically, if not self.test_suite: return - test_case = junit_xml.TestCase(name, classname=self._class_name(test_class), - elapsed_sec=duration, stdout=stdout, stderr=stderr) - if failure: - test_case.add_failure_info(failure, output=output) - self.test_suite.test_cases.append(test_case) + with self._lock: + test_case = junit_xml.TestCase(name, classname=self._class_name(test_class), + elapsed_sec=duration, stdout=stdout, stderr=stderr) + if failure: + test_case.add_failure_info(failure, output=output) + self.test_suite.test_cases.append(test_case) - self._write_test_file() + self._write_test_file() def _write_test_file(self): """Write test results to file""" - with open('nlt-junit.xml', 'w') as file: - junit_xml.TestSuite.to_file(file, [self.test_suite], prettyprint=True) + with self._lock: + with open('nlt-junit.xml', 'w') as file: + junit_xml.TestSuite.to_file(file, [self.test_suite], prettyprint=True) def explain(self, line, log_file, esignal): """Log an error, along with the other errors it caused @@ -341,29 +382,30 @@ def add(self, line, sev, message, cat=None, mtype=None): Describe an error and add it to the issues array. Add it to the pending array, for later clarification """ - entry = {} - entry['fileName'] = line.filename - if mtype: - entry['type'] = mtype - else: - entry['type'] = message - if cat: - entry['category'] = cat - entry['lineStart'] = line.lineno - # Jenkins no longer seems to display the description. - entry['description'] = message - entry['message'] = f'{line.get_anon_msg()}\n{message}' - entry['severity'] = sev - self.issues.append(entry) - if self.pending and self.pending[0][0].pid != line.pid: - self.reset_pending() - self.pending.append((line, message)) - self._flush() - if self.post or (self.post_error and sev in ('HIGH', 'ERROR')): - # https://docs.github.com/en/actions/reference/workflow-commands-for-github-actions - if self.post_error: - message = line.get_msg() - print(f'::warning file={line.filename},line={line.lineno},::{self.check}, {message}') + with self._lock: + entry = {} + entry['fileName'] = line.filename + if mtype: + entry['type'] = mtype + else: + entry['type'] = message + if cat: + entry['category'] = cat + entry['lineStart'] = line.lineno + # Jenkins no longer seems to display the description. + entry['description'] = message + entry['message'] = f'{line.get_anon_msg()}\n{message}' + entry['severity'] = sev + self.issues.append(entry) + if self.pending and self.pending[0][0].pid != line.pid: + self.reset_pending() + self.pending.append((line, message)) + self._flush() + if self.post or (self.post_error and sev in ('HIGH', 'ERROR')): + # https://docs.github.com/en/actions/reference/workflow-commands-for-github-actions + if self.post_error: + message = line.get_msg() + print(f'::warning file={line.filename},line={line.lineno},::{self.check}, {message}') def reset_pending(self): """Reset the pending list @@ -380,35 +422,37 @@ def _flush(self): from the __del__ method of DaosServer, so do not use __file__ here either. """ - self._fd.seek(0) - self._fd.truncate(0) - data = {} - data['issues'] = list(self.issues) - if self._running: - # When the test is running insert an error in case of abnormal - # exit, so that crashes in this code can be identified. - entry = {} - entry['fileName'] = self._file - # pylint: disable=protected-access - entry['lineStart'] = sys._getframe().f_lineno - entry['severity'] = 'ERROR' - entry['message'] = 'Tests are still running' - data['issues'].append(entry) - json.dump(data, self._fd, indent=2) - self._fd.flush() + with self._lock: + self._fd.seek(0) + self._fd.truncate(0) + data = {} + data['issues'] = list(self.issues) + if self._running: + # When the test is running insert an error in case of abnormal + # exit, so that crashes in this code can be identified. + entry = {} + entry['fileName'] = self._file + # pylint: disable=protected-access + entry['lineStart'] = sys._getframe().f_lineno + entry['severity'] = 'ERROR' + entry['message'] = 'Tests are still running' + data['issues'].append(entry) + json.dump(data, self._fd, indent=2) + self._fd.flush() def close(self): """Save, and close the log file""" - self._running = False - self._flush() - self._fd.close() - self._fd = None - print(f'Closed JSON file {self.filename} with {len(self.issues)} errors') - if self.test_suite: - # This is a controlled shutdown, so wipe the error saying forced exit. - self.test_suite.test_cases[1].errors = [] - self.test_suite.test_cases[1].error_message = [] - self._write_test_file() + with self._lock: + self._running = False + self._flush() + self._fd.close() + self._fd = None + print(f'Closed JSON file {self.filename} with {len(self.issues)} errors') + if self.test_suite: + # This is a controlled shutdown, so wipe the error saying forced exit. + self.test_suite.test_cases[1].errors = [] + self.test_suite.test_cases[1].error_message = [] + self._write_test_file() def load_conf(args): @@ -883,8 +927,9 @@ def _start(self): self.fetch_pools() def _stop_agent(self): - self._agent.send_signal(signal.SIGINT) - ret = self._agent.wait(timeout=5) + ret, _ = terminate_process(self._agent, 'daos_agent', graceful_signal=signal.SIGINT) + if ret is None: + ret = -1 print(f'rc from agent is {ret}') self._agent = None try: @@ -969,14 +1014,22 @@ def _stop(self, wf): self._add_test_case('stop', duration=duration) print(f'Server stopped in {duration:.2f} seconds') - self._sp.send_signal(signal.SIGTERM) - ret = self._sp.wait(timeout=5) + ret, _ = terminate_process(self._sp, 'daos_server', graceful_signal=signal.SIGTERM) + if ret is None: + ret = -1 + entry = {} + entry['fileName'] = self._file + # pylint: disable=protected-access + entry['lineStart'] = sys._getframe().f_lineno + entry['severity'] = 'ERROR' + entry['message'] = 'Unable to terminate daos_server process cleanly' + self.conf.wf.issues.append(entry) print(f'rc from server is {ret}') self.conf.compress_file(self.agent_log.name) self.conf.compress_file(self.control_log.name) - for log in self.server_logs: + for log in list(self.server_logs): log_test(self.conf, log.name, leak_wf=wf, skip_fi=self._fi) self.server_logs.remove(log) self.running = False @@ -1051,12 +1104,14 @@ def get_test_pool_obj(self): return self.test_pool - def run_daos_client_cmd(self, cmd): + def run_daos_client_cmd(self, cmd, timeout=None): """Run a DAOS client Run a command, returning what subprocess.run() would. Enable logging, and valgrind for the command. + + timeout: optional wall-clock limit in seconds; raises subprocess.TimeoutExpired if exceeded. """ valgrind_hdl = ValgrindHelper(self.conf) @@ -1082,7 +1137,7 @@ def run_daos_client_cmd(self, cmd): cmd_env['DAOS_AGENT_DRPC_DIR'] = self.conf.agent_dir rc = subprocess.run(exec_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, - env=cmd_env, check=False) + env=cmd_env, check=False, timeout=timeout) if rc.stderr != b'': print('Stderr from command') @@ -1112,7 +1167,8 @@ def run_daos_client_cmd(self, cmd): rc.returncode = 0 assert rc.returncode == 0, rc - def run_daos_client_cmd_pil4dfs(self, cmd, check=True, container=None, report=True): + def run_daos_client_cmd_pil4dfs(self, cmd, check=True, container=None, report=True, + timeout=None): """Run a DAOS client with libpil4dfs.so Run a command, returning what subprocess.run() would. @@ -1123,6 +1179,8 @@ def run_daos_client_cmd_pil4dfs(self, cmd, check=True, container=None, report=Tr Looks like valgrind and libpil4dfs.so do not work together sometime. Disable valgrind at this moment. Will revisit this issue later. + + timeout: optional wall-clock limit in seconds; raises subprocess.TimeoutExpired if exceeded. """ if container is not None: assert isinstance(container, DaosCont) @@ -1159,7 +1217,7 @@ def run_daos_client_cmd_pil4dfs(self, cmd, check=True, container=None, report=Tr print('Run command: ') print(cmd) rc = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=cwd, - env=cmd_env, check=False) + env=cmd_env, check=False, timeout=timeout) print(rc) if rc.stderr != b'': @@ -1487,8 +1545,11 @@ def start(self, v_hint=None, use_oopt=False): total_time += 1 if total_time > 60: # Kill the unresponsive dfuse command - self._sp.send_signal(signal.SIGTERM) + terminate_process(self._sp, 'dfuse(start)', graceful_signal=signal.SIGTERM) self._sp = None + umount(self.dir, background=True) + time.sleep(1) + umount(self.dir) raise NLTestFail('Timeout starting dfuse') self._daos.add_fuse(self) @@ -1538,7 +1599,7 @@ def stop(self, ignore_einval=False): fatal_errors = True except subprocess.TimeoutExpired: print('Timeout stopping dfuse') - self._sp.send_signal(signal.SIGTERM) + terminate_process(self._sp, 'dfuse(stop)', graceful_signal=signal.SIGTERM) fatal_errors = True run_leak_test = False self._sp = None @@ -1547,7 +1608,11 @@ def stop(self, ignore_einval=False): # Finally, modify the valgrind xml file to remove the # prefix to the src dir. self.valgrind.convert_xml() - os.rmdir(self.dir) + try: + os.rmdir(self.dir) + except OSError as error: + print(f'Failed to remove dfuse dir {self.dir}: {error}') + fatal_errors = True self._daos.remove_fuse(self) return fatal_errors @@ -1730,12 +1795,15 @@ def run_daos_cmd(conf, log_check=True, ignore_busy=False, use_json=False, - cwd=None): + cwd=None, + timeout=None): """Run a DAOS command Run a command, returning what subprocess.run() would. Enable logging, and valgrind for the command. + + timeout: optional wall-clock limit in seconds; raises subprocess.TimeoutExpired if exceeded. """ dcr = DaosCmdReturn() valgrind_hdl = ValgrindHelper(conf) @@ -1778,7 +1846,7 @@ def run_daos_cmd(conf, cmd_env['DAOS_AGENT_DRPC_DIR'] = conf.agent_dir rc = subprocess.run(exec_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, - env=cmd_env, check=False, cwd=cwd) + env=cmd_env, check=False, cwd=cwd, timeout=timeout) if rc.stderr != b'': print('Stderr from command') @@ -5616,6 +5684,12 @@ def test_pydaos_kv_obj_class(server, conf): # +# Maximum wall-clock seconds a single fault-injection child is allowed to run. +# If a child exceeds this limit it is forcibly terminated so one stuck process +# cannot stall the entire CI stage. +_CHILD_TIMEOUT = 120 + + class AllocFailTestRun(): """Class to run a fault injection command with a single fault""" @@ -5645,6 +5719,7 @@ def __init__(self, aft, cmd, env, loc, cwd): self._stderr = None self._fi_loc = None self._cwd = cwd + self._start_time = None if loc: prefix = f'dnt_{loc:04d}_' @@ -5715,6 +5790,7 @@ def start(self): stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE) + self._start_time = time.monotonic() def has_finished(self): """Check if the command has completed""" @@ -5723,6 +5799,14 @@ def has_finished(self): rc = self._sp.poll() if rc is None: + elapsed = time.monotonic() - self._start_time + if elapsed > _CHILD_TIMEOUT: + cmd_text = ' '.join(self._cmd) + print(f'\nFault injection child timed out after {elapsed:.0f}s ' + f'(loc={self.loc}, cmd={cmd_text!r}); terminating') + terminate_process(self._sp, cmd_text) + self._post(-signal.SIGKILL) + return True return False self._post(rc) return True @@ -5732,7 +5816,14 @@ def wait(self): if self.returncode is not None: return - self._post(self._sp.wait()) + try: + self._post(self._sp.wait(timeout=_CHILD_TIMEOUT)) + except subprocess.TimeoutExpired: + cmd_text = ' '.join(self._cmd) + print(f'\nFault injection child timed out after {_CHILD_TIMEOUT}s ' + f'(loc={self.loc}, cmd={cmd_text!r}); terminating') + terminate_process(self._sp, cmd_text) + self._post(-signal.SIGKILL) def _post(self, rc): """Helper function, called once after command is complete. @@ -6559,26 +6650,30 @@ def server_fi(args): conf.set_args(args) setup_log_test(conf) - with DaosServer(conf, wf=wf, test_class='server-fi', enable_fi=True) as server: - - pool = server.get_test_pool_obj() - cont = create_cont(conf, pool=pool, ctype='POSIX', label='server_test') - - # Instruct the server to fail a % of allocations. - server.set_fi(probability=1) - - for idx in range(100): - server.run_daos_client_cmd_pil4dfs( - ['touch', f'file.{idx}'], container=cont, check=False, report=False) - server.run_daos_client_cmd_pil4dfs( - ['dd', 'if=/dev/zero', f'of=file.{idx}', 'bs=1', 'count=1024'], - container=cont, check=False, report=False) - server.run_daos_client_cmd_pil4dfs( - ['rm', '-f', f'file.{idx}'], container=cont, check=False, report=False) - - # Turn off fault injection again to assist in server shutdown. - server.set_fi(probability=0) - server.set_fi(probability=0) + try: + with DaosServer(conf, wf=wf, test_class='server-fi', enable_fi=True) as server: + + pool = server.get_test_pool_obj() + cont = create_cont(conf, pool=pool, ctype='POSIX', label='server_test') + + # Instruct the server to fail a % of allocations. + server.set_fi(probability=1) + + for idx in range(100): + server.run_daos_client_cmd_pil4dfs( + ['touch', f'file.{idx}'], container=cont, check=False, report=False) + server.run_daos_client_cmd_pil4dfs( + ['dd', 'if=/dev/zero', f'of=file.{idx}', 'bs=1', 'count=1024'], + container=cont, check=False, report=False) + server.run_daos_client_cmd_pil4dfs( + ['rm', '-f', f'file.{idx}'], container=cont, check=False, report=False) + + # Turn off fault injection again to assist in server shutdown. + server.set_fi(probability=0) + server.set_fi(probability=0) + finally: + wf.close() + close_log_test(conf) def generate_special_test_list(): @@ -6893,6 +6988,105 @@ def _positive_int(value): return ivalue +def _load_warning_issues(filename): + """Load a warnings json file and return issues + parse error string if any.""" + if not os.path.exists(filename): + return ([], None) + + try: + with open(filename, 'r') as infile: + data = json.load(infile) + except (OSError, ValueError, TypeError) as error: + return ([], f'Unable to parse {filename}: {error}') + + issues = data.get('issues', []) + if not isinstance(issues, list): + return ([], f'Invalid issues format in {filename}') + return (issues, None) + + +def _count_severity(issues): + counts = {'LOW': 0, 'NORMAL': 0, 'HIGH': 0, 'ERROR': 0} + for issue in issues: + sev = issue.get('severity', 'NORMAL') + counts[sev] = counts.get(sev, 0) + 1 + return counts + + +def _collect_high_findings(filename, issues): + findings = [] + for issue in issues: + sev = issue.get('severity') + if sev not in ('HIGH', 'ERROR'): + continue + findings.append({ + 'source': filename, + 'severity': sev, + 'file': issue.get('fileName'), + 'line': issue.get('lineStart'), + 'message': issue.get('message') + }) + findings.sort(key=lambda item: 0 if item['severity'] == 'ERROR' else 1) + return findings + + +def write_nlt_summary(args, fatal_errors=None, exception=None): + """Write a compact run summary as nlt-summary.json.""" + warning_files = ['nlt-errors.json', 'nlt-server-leaks.json', 'nlt-client-leaks.json'] + warning_summary = {} + parse_errors = [] + high_findings = [] + + for filename in warning_files: + issues, parse_error = _load_warning_issues(filename) + if parse_error: + parse_errors.append(parse_error) + warning_summary[filename] = { + 'present': os.path.exists(filename), + 'issue_count': len(issues), + 'severity_counts': _count_severity(issues) + } + high_findings.extend(_collect_high_findings(filename, issues)) + + result_failed = bool(exception) + if fatal_errors is not None: + result_failed = result_failed or bool(fatal_errors.errors) + + summary = { + 'schema_version': 1, + 'generated_at_utc': time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime()), + 'result': 'failed' if result_failed else 'passed', + 'run': { + 'mode': args.mode, + 'class_name': args.class_name, + 'repeat': args.repeat, + 'engine_count': args.engine_count, + 'hostname': socket.gethostname(), + 'jenkins_node': os.environ.get('NODE_NAME') + }, + 'warnings': warning_summary, + 'high_severity_findings': high_findings[:20], + 'artifacts': { + 'nlt-junit.xml': os.path.exists('nlt-junit.xml'), + 'nlt_logs': os.path.exists('nlt_logs'), + 'nltir.xml': os.path.exists('nltir.xml'), + 'nltr.json': os.path.exists('nltr.json') + } + } + + if exception: + summary['exception'] = { + 'type': exception.__class__.__name__, + 'message': str(exception) + } + + if parse_errors: + summary['parse_errors'] = parse_errors + + with open('nlt-summary.json', 'w') as outfile: + json.dump(summary, outfile, indent=2) + + def main(): """Wrap the core function, and catch/report any exceptions @@ -6939,7 +7133,14 @@ def main(): resource.setrlimit(resource.RLIMIT_NOFILE, (hard, hard)) if args.server_fi: - server_fi(args) + run_error = None + try: + server_fi(args) + except Exception as error: # pylint: disable=broad-exception-caught + run_error = error + raise + finally: + write_nlt_summary(args, exception=run_error) return if args.mode: @@ -6973,11 +7174,14 @@ def main(): class_id=args.class_name, junit=True) + fatal_errors = None + run_error = None try: fatal_errors = run(wf, args) wf.add_test_case('exit_wrapper') wf.close() except Exception as error: + run_error = error print(error) print(str(error)) print(repr(error)) @@ -6985,6 +7189,8 @@ def main(): wf.add_test_case('exit_wrapper', str(error), output=trace) wf.close() raise + finally: + write_nlt_summary(args, fatal_errors=fatal_errors, exception=run_error) if fatal_errors.errors: print("Significant errors encountered")