mirror of
https://github.com/facebook/rocksdb.git
synced 2026-11-06 02:52:35 -05:00
Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 Hui Xiao
|
f2b6fbf7b3 |
No files matched your search
@@ -0,0 +1,220 @@
|
||||
# CPU Corruption Injector
|
||||
|
||||
## Overview
|
||||
|
||||
This tool measures the **outcome distribution** over many `db_stress` test runs.
|
||||
In each `db_stress` test run, a single CPU-register bit-flip is injected into
|
||||
exactly one `db_stress` operation (a write, flush, or compaction) running under
|
||||
`gdb`. Aggregating over many such runs answers "when the CPU corrupts one
|
||||
instruction inside that op, how often does RocksDB catch it, crash, silently
|
||||
lose/return wrong data, or absorb it with no effect?".
|
||||
|
||||
It has three layers:
|
||||
|
||||
- **Detection** (`db_stress --verify_cpu_corruption_dir`): right after the
|
||||
injected op, `db_stress` reads the whole keyspace back and compares against its
|
||||
expected state, writing any finding to a `data_corruption.*.json` file.
|
||||
- **Injection** (`injector*.py`): runs inside `gdb`, navigates to a randomly
|
||||
chosen instance of the op, stops at a random "critical" instruction -- one of
|
||||
the two kinds that can carry a fault into data: a **move** carrying a value
|
||||
between CPU and memory (a general-purpose or vector register), or a
|
||||
**branch/jump** (the `eflags` register) -- and flips one random bit in the
|
||||
register it uses. Control registers (`rsp`/`rbp`/`rip`/segment) are left alone.
|
||||
- **Orchestration** (`runner*.py`): builds the run set, launches `gdb` per run,
|
||||
classifies each outcome, and aggregates them.
|
||||
|
||||
## Requirements
|
||||
|
||||
- A `db_stress` binary built with debug info and no inlining, so `gdb` can set
|
||||
breakpoints on the injection-site functions and read their source lines:
|
||||
|
||||
```bash
|
||||
make DEBUG_LEVEL=0 EXTRA_CXXFLAGS="-g -fno-inline" db_stress
|
||||
```
|
||||
|
||||
- `gdb`. Defaults to `/usr/bin/gdb`; override with the `ICC_GDB` env var if the
|
||||
telemetry's source resolves to `@ ?:0` (a `gdb` too old to read this build's
|
||||
debug info).
|
||||
- Optional per-run timeout via the `ICC_RUN_TIMEOUT` env var (seconds, default
|
||||
`600`). One hung `gdb`+`db_stress` cannot block the rest of the set.
|
||||
|
||||
## Entry point: `runner.py`
|
||||
|
||||
A **campaign** is many `db_stress` test **runs** of one op type. Each run is a
|
||||
full single-threaded `db_stress` test that performs `ops_per_thread` ops, but the
|
||||
injector corrupts exactly **one** op instance of the chosen type in that run
|
||||
(e.g. the 51st Put of that run); the run is then classified into a single outcome
|
||||
bucket.
|
||||
|
||||
- `--op` is one of `write`, `flush`, `compaction`.
|
||||
- `--parallel N` runs `N` runs at once (each `db_stress` run is itself
|
||||
single-threaded, so raise this to use more cores).
|
||||
- `--seed N` reproduces a set: run `i` uses seed `N+i`. With `--seed 0` (the
|
||||
default) a fresh random `base_seed` is chosen and logged; pass that value back
|
||||
via `--seed` to replay, and add `--runs 1` with `--seed=<base_seed+i>` to
|
||||
replay a single run `i`.
|
||||
|
||||
```bash
|
||||
# a compaction campaign of 100 runs
|
||||
python3 tools/cpu_corruption_injector/runner.py --op compaction --runs 100 \
|
||||
--stress_cmd <db_stress> --report_dir <dir>
|
||||
|
||||
# a flush campaign of 200 runs, 8 in parallel
|
||||
python3 tools/cpu_corruption_injector/runner.py --op flush --runs 200 \
|
||||
--stress_cmd <db_stress> --report_dir <dir> --parallel 8
|
||||
|
||||
# reproduce an earlier set from its logged base_seed
|
||||
python3 tools/cpu_corruption_injector/runner.py --op write --runs 100 \
|
||||
--stress_cmd <db_stress> --report_dir <dir> --seed 12345678
|
||||
```
|
||||
|
||||
Before doing any work, `runner.py` preflights the build: it confirms `gdb` can
|
||||
set breakpoints on the op's injection-site functions and read their source
|
||||
lines, and fails fast otherwise.
|
||||
|
||||
## Outputs
|
||||
|
||||
Everything lands under `--report_dir`: two campaign-wide files at the top level,
|
||||
plus one `run_XXXXX/` subdirectory per run.
|
||||
|
||||
```text
|
||||
<report_dir>/
|
||||
├── summary.json # campaign-wide: outcome_counts + a record per run
|
||||
├── runner.log # campaign-wide: runner logging (also echoed to console)
|
||||
├── run_00000/ # one subdirectory per run
|
||||
│ ├── inject.json # what the injector did (WHERE / WHAT / HOW)
|
||||
│ ├── data_corruption.<tid>.json # the finding, only if this run flagged an SDC/corruption
|
||||
│ ├── gdb.log # this run's gdb output
|
||||
│ ├── db/ # this run's db_stress DB (working dir)
|
||||
│ └── exp/ # this run's expected-values (working dir)
|
||||
├── run_00001/
|
||||
│ └── ...
|
||||
└── ...
|
||||
```
|
||||
|
||||
So `summary.json` and `runner.log` are at the top of `--report_dir`; the
|
||||
per-run `gdb.log` / `inject.json` / `data_corruption.*.json` live inside each
|
||||
`run_XXXXX/`.
|
||||
|
||||
Outcome buckets:
|
||||
|
||||
- **SDC**: silent data corruption -- the read-back succeeded but returned the
|
||||
wrong result (a lost key, a resurrected key, or a wrong value).
|
||||
- **CORRUPTION**: `db_stress` detected a corruption via an integrity check
|
||||
(`Status::Corruption`).
|
||||
- **CRASH**: the injected corruption crashed `db_stress` (a fatal signal).
|
||||
- **NO_EFFECT**: the corruption landed but the run finished with correct data.
|
||||
- **NO_INJECTION**: no bit was flipped this run. The op instance, its
|
||||
`target_fn`, and the instruction to corrupt are all picked **randomly** up
|
||||
front, so a run can occasionally target something it never reaches -- a random
|
||||
op instance beyond the ops this run actually performed, or a
|
||||
critical-instruction index that no later `target_fn` call happens to have.
|
||||
Should be rare; a high count means the injection sites (or the random ranges)
|
||||
need revisiting.
|
||||
- **ERROR**: a harness/framework failure, **not** a `db_stress` data outcome --
|
||||
the injector hit a bug, `gdb` died before writing `inject.json`, or the run hit
|
||||
the `ICC_RUN_TIMEOUT` cap. Expected to be **0**; any non-zero count is a
|
||||
framework bug worth investigating.
|
||||
|
||||
## Reading a run's output across the 3 layers
|
||||
|
||||
Each run threads through all three layers: the injector records what it corrupted
|
||||
(`inject.json`), `db_stress` records what the post-op read-back found
|
||||
(`data_corruption.<tid>.json`), and the runner rolls every run up into
|
||||
`summary.json`. The examples below are from a real 40-run compaction campaign:
|
||||
|
||||
```bash
|
||||
python3 tools/cpu_corruption_injector/runner.py --op compaction --runs 40 \
|
||||
--seed 424242 --stress_cmd <db_stress> --report_dir <dir>
|
||||
```
|
||||
|
||||
**Orchestration layer** -- `summary.json`'s `outcome_counts` (how many of the
|
||||
campaign's runs landed in each bucket), the whole campaign at a glance:
|
||||
|
||||
| SDC | CORRUPTION | CRASH | NO_EFFECT | NO_INJECTION | ERROR |
|
||||
| --- | ---------- | ----- | --------- | ------------ | ----- |
|
||||
| 2 | 10 | 0 | 25 | 3 | 0 |
|
||||
|
||||
### A silent data corruption (SDC): `run_00023`
|
||||
|
||||
**Injection layer** -- `run_00023/inject.json`: an `eflags` flag-flip on the
|
||||
compaction iterator's key-copy path; the op did NOT crash
|
||||
(`db_stress_crash_signal: null`):
|
||||
|
||||
```json
|
||||
{
|
||||
"injection_result": "injected",
|
||||
"db_stress_crash_signal": null,
|
||||
"op": "compaction",
|
||||
"op_index": 6,
|
||||
"entry_fn": "rocksdb::CompactionJob::Run",
|
||||
"target_fn": "rocksdb::CompactionIterator::NextFromInput",
|
||||
"critical_instruction_index": 0,
|
||||
"corruptions": [
|
||||
{
|
||||
"instruction": "cmp $0x40,%rdx",
|
||||
"register": "eflags",
|
||||
"corruption_type": "flag_flip",
|
||||
"before": "0x287",
|
||||
"after": "0x286",
|
||||
"details": {
|
||||
"source": "__memmove_avx512_unaligned_erms @ ../sysdeps/x86_64/multiarch/memmove-vec-unaligned-erms.S:299",
|
||||
"call_chain": [
|
||||
"__memmove_avx512_unaligned_erms @ ../sysdeps/x86_64/multiarch/memmove-vec-unaligned-erms.S:299",
|
||||
"rocksdb::IterKey::SetKeyImpl @ ./db/dbformat.h:941",
|
||||
"rocksdb::IterKey::SetInternalKey @ ./db/dbformat.h:741",
|
||||
"rocksdb::IterKey::SetInternalKey @ ./db/dbformat.h:749",
|
||||
"rocksdb::CompactionIterator::NextFromInput @ db/compaction/compaction_iterator.cc:781"
|
||||
]
|
||||
}
|
||||
}
|
||||
],
|
||||
"ops_seen": 6,
|
||||
"critical_instructions_seen": 1
|
||||
}
|
||||
```
|
||||
|
||||
**Detection layer** -- `run_00023/data_corruption.<tid>.json`: the post-op
|
||||
read-back found a committed key missing:
|
||||
|
||||
```json
|
||||
{
|
||||
"kind": "lost",
|
||||
"cf": 0,
|
||||
"key": 202,
|
||||
"value_from_db": "",
|
||||
"value_from_expected": "010000000504070609080B0A0D0C0F0E",
|
||||
"op_status": "Get: NotFound"
|
||||
}
|
||||
```
|
||||
|
||||
Reading it: the flag flip mis-copied the compaction output key, dropping key 202
|
||||
from the output SST. The op returned OK and no checksum fired, but reading key
|
||||
202 back returns `NotFound` for a committed key, so the runner classifies the run
|
||||
`SDC` (`kind=lost`) -- the bug class this tool exists to surface.
|
||||
|
||||
## Auxiliary entry point: `injector.py`
|
||||
|
||||
`injector.py` is **not run on its own** -- there is no `python3 injector.py`
|
||||
usage. It only works inside `gdb`'s embedded Python (it does `import gdb`), so
|
||||
`runner.py` runs it by launching `gdb` with the injector as gdb's script. Each
|
||||
run's invocation looks like:
|
||||
|
||||
```bash
|
||||
gdb --batch --nx \
|
||||
-iex 'py import sys; sys.argv=[...]' \
|
||||
-x tools/cpu_corruption_injector/injector.py \
|
||||
--args <db_stress> <db_stress flags...>
|
||||
```
|
||||
|
||||
- `--batch` runs `gdb` non-interactively and exits when the script finishes;
|
||||
`--nx` skips the user's `~/.gdbinit` so their config can't change behavior.
|
||||
- `-iex 'py ... sys.argv=[...]'` runs a Python command *before* the script loads,
|
||||
seeding `injector.py`'s `sys.argv` with its parameters (`--op`, `--op_index`,
|
||||
`--entry_fn`, `--target_fn`, `--seed`, `--dir`) -- gdb otherwise hands a `-x`
|
||||
script an empty argv.
|
||||
- `-x tools/cpu_corruption_injector/injector.py` is the script `gdb` executes in
|
||||
its embedded Python.
|
||||
- `--args <db_stress> <db_stress flags...>` is the program `gdb` launches and
|
||||
debugs -- the `db_stress` binary and its flags, which `gdb` starts, stops
|
||||
mid-op, and corrupts.
|
||||
@@ -0,0 +1,216 @@
|
||||
#!/usr/bin/env python3
|
||||
# Copyright (c) Meta Platforms, Inc. and affiliates.
|
||||
# This source code is licensed under both the GPLv2 (found in the COPYING file in the root directory)
|
||||
# and the Apache 2.0 License (found in the LICENSE.Apache file in the root directory).
|
||||
|
||||
# pyre-strict
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import logging
|
||||
import os
|
||||
import random
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
from concurrent.futures import as_completed, ThreadPoolExecutor
|
||||
|
||||
# Run as a plain script (not an installed package), so this directory is not on the
|
||||
# import path; add it so the sibling runner_* imports below resolve (hence each E402).
|
||||
_HERE = os.path.dirname(os.path.abspath(__file__))
|
||||
if _HERE not in sys.path:
|
||||
sys.path.append(_HERE)
|
||||
|
||||
from runner_execute import _get_gdb, execute # noqa: E402
|
||||
from runner_build import build_runs, INJECTION_SITES, Run # noqa: E402
|
||||
from runner_report import report # noqa: E402
|
||||
|
||||
logger: logging.Logger = logging.getLogger("runner")
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
args = parse_args(argv)
|
||||
report_dir = os.path.abspath(args.report_dir)
|
||||
stress_cmd = os.path.abspath(args.stress_cmd)
|
||||
parallel_runs = max(args.parallel, 1)
|
||||
base_seed = args.seed or random.randint(1, 2**64) # 0 -> a fresh seed each run
|
||||
|
||||
os.makedirs(report_dir, exist_ok=True)
|
||||
_setup_logging(report_dir)
|
||||
|
||||
# Fail fast on the deterministic, whole-run problems before doing any work.
|
||||
if not os.path.exists(stress_cmd):
|
||||
logger.error("stress_cmd not found at %s", stress_cmd)
|
||||
return 2
|
||||
try:
|
||||
verify_injection_site(stress_cmd, args.op)
|
||||
except RuntimeError as e:
|
||||
logger.error("%s", e)
|
||||
return 2
|
||||
|
||||
logger.info(
|
||||
"runner: op=%s runs=%d base_seed=%d report_dir=%s parallel=%d gdb=%s",
|
||||
args.op,
|
||||
args.runs,
|
||||
base_seed,
|
||||
report_dir,
|
||||
parallel_runs,
|
||||
_get_gdb(),
|
||||
)
|
||||
# One number reproduces everything: run i used seed base_seed + i, so the whole set
|
||||
# replays with --seed=base_seed, and a single run i with --seed=<base_seed+i> --runs 1.
|
||||
logger.info("reproduce with --seed=%d (run i used seed %d+i)", base_seed, base_seed)
|
||||
injector_py = os.path.join(_HERE, "injector.py")
|
||||
runs = build_runs(args.op, args.runs, report_dir, base_seed)
|
||||
execute_all(runs, injector_py, stress_cmd, parallel_runs)
|
||||
report(runs, args.op, report_dir, base_seed)
|
||||
return 0
|
||||
|
||||
|
||||
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser(
|
||||
description="CPU corruption injection runner: runs db_stress under gdb and, "
|
||||
"per run, injects CPU corruption into a single db_stress operation (a write, "
|
||||
"flush, or compaction), then classifies the outcome."
|
||||
)
|
||||
parser.add_argument(
|
||||
"--op", required=True, choices=tuple(INJECTION_SITES), help="op type to inject"
|
||||
)
|
||||
parser.add_argument(
|
||||
"--runs",
|
||||
type=int,
|
||||
required=True,
|
||||
help="how many runs to do (each run injects into one op).",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--stress_cmd", required=True, help="path to the db_stress binary"
|
||||
)
|
||||
parser.add_argument(
|
||||
"--report_dir",
|
||||
required=True,
|
||||
help="output dir for ALL artifacts: runner.log, summary.json, and one "
|
||||
"run_XXXXX/ per run (gdb.log, inject.json, data_corruption.*.json).",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--parallel",
|
||||
type=int,
|
||||
default=1,
|
||||
help="how many runs to execute at once (default 1); each db_stress run is "
|
||||
"itself single-threaded, so raise this to use more cores.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--seed",
|
||||
type=int,
|
||||
default=0,
|
||||
help="master seed for the run set. 0 (default) picks a fresh random seed each "
|
||||
"invocation (logged as base_seed=N); pass that N back via --seed to reproduce "
|
||||
"the same runs. Run i uses seed N+i.",
|
||||
)
|
||||
return parser.parse_args(argv)
|
||||
|
||||
|
||||
def _setup_logging(report_dir: str) -> None:
|
||||
# Log to the console AND <report_dir>/runner.log, so runner-level logging lands
|
||||
# beside the per-run artifacts and the summary.
|
||||
fmt = logging.Formatter("%(asctime)s %(levelname)s %(message)s")
|
||||
console = logging.StreamHandler()
|
||||
console.setFormatter(fmt)
|
||||
log_file = logging.FileHandler(os.path.join(report_dir, "runner.log"))
|
||||
log_file.setFormatter(fmt)
|
||||
root_logger = logging.getLogger()
|
||||
root_logger.setLevel(logging.INFO)
|
||||
root_logger.handlers = [console, log_file]
|
||||
|
||||
|
||||
def verify_injection_site(stress_cmd: str, op: str) -> None:
|
||||
# Before running anything, confirm gdb can work with this op's injection site on this
|
||||
# db_stress build: every function we break on still resolves, and gdb can read its
|
||||
# source file:line. Checking entry_fn's source line alone catches a gdb too old to
|
||||
# read this build's debug info (which makes the telemetry's source come out "@ ?:0").
|
||||
site = INJECTION_SITES[op]
|
||||
functions = list(dict.fromkeys([site.entry_fn, *site.target_fns]))
|
||||
gdb_output = _run_gdb_check(stress_cmd, functions, site.entry_fn)
|
||||
|
||||
missing = [fn for fn in functions if _function_not_found(gdb_output, fn)]
|
||||
if missing:
|
||||
raise RuntimeError(
|
||||
f"gdb could not set a breakpoint on these functions in "
|
||||
f"{os.path.basename(stress_cmd)} (renamed, fully inlined, or not in this "
|
||||
f"build?): {', '.join(missing)}\n{gdb_output}"
|
||||
)
|
||||
if _no_source_lines(gdb_output):
|
||||
raise RuntimeError(
|
||||
f"gdb resolved the functions but cannot read their source line numbers; "
|
||||
f"this gdb ({_get_gdb()}) is likely too old for this build -- point ICC_GDB at a "
|
||||
f"newer one.\n{gdb_output}"
|
||||
)
|
||||
logger.info("gdb check OK for op=%s: %s", op, ", ".join(functions))
|
||||
|
||||
|
||||
def _run_gdb_check(stress_cmd: str, functions: list[str], entry_fn: str) -> str:
|
||||
# gdb loads db_stress only for its symbols (it never runs it, so no db_stress flags
|
||||
# are needed): break on each function, ask entry_fn's source line, then quit. --nx
|
||||
# ignores the user's ~/.gdbinit so their gdb config can't change the output.
|
||||
commands: list[str] = []
|
||||
for function in functions:
|
||||
commands += ["-ex", f"break {function}"]
|
||||
commands += ["-ex", f"info line {entry_fn}", "-ex", "quit"]
|
||||
try:
|
||||
finished = subprocess.run(
|
||||
[_get_gdb(), "--batch", "--nx", *commands, stress_cmd],
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.STDOUT,
|
||||
timeout=180,
|
||||
check=False,
|
||||
)
|
||||
except subprocess.TimeoutExpired as e:
|
||||
raise RuntimeError(f"gdb check timed out: {e}") from e
|
||||
except OSError as e:
|
||||
raise RuntimeError(f"could not launch gdb ({_get_gdb()}): {e}") from e
|
||||
return finished.stdout.decode("utf-8", "replace")
|
||||
|
||||
|
||||
def _function_not_found(gdb_output: str, function: str) -> bool:
|
||||
short_name = function.rsplit("::", 1)[-1] # "rocksdb::MemTable::Add" -> "Add"
|
||||
markers = (
|
||||
f'Function "{function}" not defined',
|
||||
f'Function "{short_name}" not defined',
|
||||
f'No symbol "{function}"',
|
||||
f'No symbol "{short_name}"',
|
||||
)
|
||||
return any(marker in gdb_output for marker in markers)
|
||||
|
||||
|
||||
def _no_source_lines(gdb_output: str) -> bool:
|
||||
# A gdb too old for this build's debug info resolves no source line (the telemetry's
|
||||
# source would come out "@ ?:0"): a Dwarf Error, or `info line` yielding no "Line N
|
||||
# of ..." at all. A function's cold-split (.cold) range legitimately has no line info,
|
||||
# so that benign "No line number information" message alone is not a failure.
|
||||
if "Dwarf Error" in gdb_output:
|
||||
return True
|
||||
return re.search(r"Line \d+ of ", gdb_output) is None
|
||||
|
||||
|
||||
def execute_all(
|
||||
runs: list[Run],
|
||||
injector_py: str,
|
||||
stress_cmd: str,
|
||||
parallel_runs: int,
|
||||
) -> None:
|
||||
with ThreadPoolExecutor(max_workers=parallel_runs) as pool:
|
||||
runs_by_future = {
|
||||
pool.submit(execute, run, injector_py, stress_cmd): run for run in runs
|
||||
}
|
||||
for future in as_completed(runs_by_future):
|
||||
run = runs_by_future[future]
|
||||
try:
|
||||
future.result()
|
||||
except Exception: # noqa: BLE001 -- one run's crash is its own ERROR, never aborts the rest
|
||||
run.outcome = "ERROR"
|
||||
logger.exception("run %d crashed", run.index)
|
||||
logger.info("run %d: %s", run.index, run.outcome)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,236 @@
|
||||
#!/usr/bin/env python3
|
||||
# Copyright (c) Meta Platforms, Inc. and affiliates.
|
||||
# This source code is licensed under both the GPLv2 (found in the COPYING file in the root directory)
|
||||
# and the Apache 2.0 License (found in the LICENSE.Apache file in the root directory).
|
||||
|
||||
# pyre-strict
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import random
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Optional
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class InjectionSite:
|
||||
# entry_fn runs once per op; we stop on its op_index-th call to pick the op to
|
||||
# corrupt. We do not single-step entry_fn itself (for flush/compaction it is huge);
|
||||
# target_fns are the smaller inner functions we single-step inside that op to find a
|
||||
# critical instruction to corrupt (one target_fn is chosen per run).
|
||||
entry_fn: str
|
||||
target_fns: list[str]
|
||||
|
||||
|
||||
INJECTION_SITES: dict[str, InjectionSite] = {
|
||||
"write": InjectionSite(
|
||||
entry_fn="rocksdb::MemTable::Add",
|
||||
target_fns=["rocksdb::MemTable::Add"],
|
||||
),
|
||||
"flush": InjectionSite(
|
||||
entry_fn="rocksdb::FlushJob::Run",
|
||||
target_fns=[
|
||||
"rocksdb::CompactionIterator::NextFromInput",
|
||||
"rocksdb::BlockBuilder::Add",
|
||||
],
|
||||
),
|
||||
"compaction": InjectionSite(
|
||||
entry_fn="rocksdb::CompactionJob::Run",
|
||||
target_fns=[
|
||||
"rocksdb::CompactionIterator::NextFromInput",
|
||||
"rocksdb::BlockBuilder::Add",
|
||||
],
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
# The complete db_stress config every run uses (the groups below merged). Together they
|
||||
# keep one injected CPU corruption isolated and observable, so they are load-bearing for
|
||||
# both correctness and detection effectiveness -- changing them weakens the signal.
|
||||
|
||||
# A mixed write/delete workload; reads and iterators are off so a single injected
|
||||
# corruption stays visible across the following ops.
|
||||
_MIXED_WORKLOAD: dict[str, object] = {
|
||||
"ops_per_thread": 1000,
|
||||
"max_key": 1000,
|
||||
"value_size_mult": 8,
|
||||
"column_families": 1,
|
||||
"nooverwritepercent": 0,
|
||||
"writepercent": 50,
|
||||
"delpercent": 40,
|
||||
"delrangepercent": 10,
|
||||
"readpercent": 0,
|
||||
"iterpercent": 0,
|
||||
"prefixpercent": 0,
|
||||
"flush_one_in": 20,
|
||||
"compact_range_one_in": 10,
|
||||
"compact_files_one_in": 10,
|
||||
# 64 KB, small on purpose: makes flush/compaction emit several SST files so there
|
||||
# are more critical instructions to corrupt.
|
||||
"target_file_size_base": 65536,
|
||||
"reopen": 0,
|
||||
"destroy_db_initially": 1,
|
||||
"clear_column_family_one_in": 0,
|
||||
"disable_wal": 0,
|
||||
}
|
||||
|
||||
# Keep the injected op the only thing running, on one thread at a time, so (1) the op +
|
||||
# its injected corruption + the post-op read-back form a clean attributable sequence, and
|
||||
# (2) no sibling thread ever sits on the single-step target breakpoint and steals gdb's
|
||||
# current frame mid-step. Each flag pins off one source of a competing thread.
|
||||
_SERIALIZED: dict[str, object] = {
|
||||
"threads": 1, # one workload thread
|
||||
"num_dbs": 1,
|
||||
"disable_auto_compactions": 1, # no surprise background compaction
|
||||
"write_buffer_size": 1 << 30, # memtable never auto-flushes
|
||||
# Compaction runs locally, not on the simulated remote compaction service's worker
|
||||
# thread (db_stress default remote_compaction_worker_threads=2). Remote runs the work
|
||||
# in CompactionServiceCompactionJob::Run, where the injector's CompactionJob::Run
|
||||
# entry_fn is off-stack, so the op is never reached. Injecting into remote compaction
|
||||
# is a planned follow-up (a second entry_fn, pinned to one worker).
|
||||
"remote_compaction_worker_threads": 0,
|
||||
"subcompactions": 1, # one compaction thread, no parallel subcompactions
|
||||
}
|
||||
|
||||
# Every integrity check on -- the high-value goal is surfacing silent data corruption
|
||||
# that would otherwise pass the read-back unnoticed.
|
||||
_SDC_PROTECTIONS: dict[str, object] = {
|
||||
"batch_protection_bytes_per_key": 8,
|
||||
"memtable_protection_bytes_per_key": 8,
|
||||
"block_protection_bytes_per_key": 8,
|
||||
"paranoid_file_checks": 1,
|
||||
"paranoid_memory_checks": 1,
|
||||
"memtable_verify_per_key_checksum_on_seek": 1,
|
||||
"detect_filter_construct_corruption": "true",
|
||||
"verify_checksum": 1,
|
||||
"compression_checksum": 1,
|
||||
"file_checksum_impl": "crc32c",
|
||||
# Extra read-time checksum verification. Caution: a "read returns OK but checksum
|
||||
# fails" outcome is confusing and can also reflect a pre-existing issue, not our
|
||||
# injection -- interpret these with care.
|
||||
"verify_checksum_one_in": 1,
|
||||
"verify_file_checksums_one_in": 1000,
|
||||
}
|
||||
|
||||
# db_stress's own fault injection off, so the injected CPU corruption is the only fault.
|
||||
_NO_OTHER_FAULTS: dict[str, object] = {
|
||||
"read_fault_one_in": 0,
|
||||
"write_fault_one_in": 0,
|
||||
"metadata_read_fault_one_in": 0,
|
||||
"metadata_write_fault_one_in": 0,
|
||||
"open_read_fault_one_in": 0,
|
||||
"open_write_fault_one_in": 0,
|
||||
"open_metadata_read_fault_one_in": 0,
|
||||
"open_metadata_write_fault_one_in": 0,
|
||||
"secondary_cache_fault_one_in": 0,
|
||||
"sync_fault_injection": 0,
|
||||
}
|
||||
|
||||
DB_STRESS_PRESET: dict[str, object] = {
|
||||
**_MIXED_WORKLOAD,
|
||||
**_SERIALIZED,
|
||||
**_SDC_PROTECTIONS,
|
||||
**_NO_OTHER_FAULTS,
|
||||
}
|
||||
|
||||
|
||||
def _int_flag(name: str) -> int:
|
||||
return int(DB_STRESS_PRESET[name]) # type: ignore[arg-type]
|
||||
|
||||
|
||||
def op_count_estimate(op: str) -> int:
|
||||
"""Rough upper bound on how many of this op a run performs; the caller uses it to
|
||||
bound which op instance to target."""
|
||||
ops = _int_flag("ops_per_thread")
|
||||
if op == "write":
|
||||
write_pct = (
|
||||
_int_flag("writepercent")
|
||||
+ _int_flag("delpercent")
|
||||
+ _int_flag("delrangepercent")
|
||||
)
|
||||
return max(ops * write_pct // 100, 1)
|
||||
if op == "flush":
|
||||
return max(ops // _int_flag("flush_one_in"), 1)
|
||||
per_range = ops // _int_flag("compact_range_one_in")
|
||||
per_files = ops // _int_flag("compact_files_one_in")
|
||||
return max(per_range + per_files, 1)
|
||||
|
||||
|
||||
@dataclass
|
||||
class Run:
|
||||
# Inputs, chosen by build_runs():
|
||||
index: int
|
||||
op: str
|
||||
op_index: int
|
||||
entry_fn: str
|
||||
target_fn: str
|
||||
seed: int
|
||||
dir: str
|
||||
stress_flags: list[str] = field(default_factory=list)
|
||||
# Outcome, set by runner_execute.execute():
|
||||
outcome: str = "" # the result bucket: SDC / CORRUPTION / CRASH / NO_EFFECT / ...
|
||||
# inject.json contents, kept because runner_report reads them for the pick spread;
|
||||
# db_stress's data_corruption.json is not kept (it only feeds `outcome`).
|
||||
inject_json: Optional[dict[str, object]] = None
|
||||
|
||||
|
||||
def build_runs(
|
||||
op: str,
|
||||
run_count: int,
|
||||
report_dir: str,
|
||||
base_seed: int,
|
||||
) -> list[Run]:
|
||||
site = INJECTION_SITES[op]
|
||||
# op_index picks which op instance the injector targets. Warmup single-steps that
|
||||
# instance's target_fn call to choose a critical-instruction index; a later instance
|
||||
# may have fewer critical instructions, so the injector retries the next instances
|
||||
# (op_index+1, +2, ...) until one reaches that index. Pick op_index in the first third
|
||||
# of the estimate so enough later instances remain to retry into.
|
||||
op_index_max = max(op_count_estimate(op) // 3, 1)
|
||||
runs: list[Run] = []
|
||||
for index in range(run_count):
|
||||
seed = (base_seed + index) % 2**64
|
||||
rng = random.Random(seed)
|
||||
op_index = rng.randint(1, op_index_max)
|
||||
target_fn = rng.choice(site.target_fns)
|
||||
run_dir = os.path.join(report_dir, f"run_{index:05d}")
|
||||
os.makedirs(run_dir, exist_ok=True)
|
||||
per_run_stress_flags = _per_run_stress_flags(run_dir, seed)
|
||||
runs.append(
|
||||
Run(
|
||||
index=index,
|
||||
op=op,
|
||||
op_index=op_index,
|
||||
entry_fn=site.entry_fn,
|
||||
target_fn=target_fn,
|
||||
seed=seed,
|
||||
dir=run_dir,
|
||||
stress_flags=_fixed_stress_flags(per_run_stress_flags),
|
||||
)
|
||||
)
|
||||
return runs
|
||||
|
||||
|
||||
def _per_run_stress_flags(run_dir: str, seed: int) -> dict[str, object]:
|
||||
# The db_stress flags unique to one run -- its own db / expected-values dirs (created
|
||||
# here), its seed, and the dir db_stress writes corruption findings to. Shared by both
|
||||
# the fixed and randomized flag paths.
|
||||
db = os.path.join(run_dir, "db")
|
||||
exp = os.path.join(run_dir, "exp")
|
||||
os.makedirs(db, exist_ok=True)
|
||||
os.makedirs(exp, exist_ok=True)
|
||||
return {
|
||||
"db": db,
|
||||
"expected_values_dir": exp,
|
||||
"seed": seed,
|
||||
"verify_cpu_corruption_dir": run_dir,
|
||||
}
|
||||
|
||||
|
||||
def _fixed_stress_flags(per_run_stress_flags: dict[str, object]) -> list[str]:
|
||||
# The fixed db_stress flag list: this run's flags plus every DB_STRESS_PRESET flag.
|
||||
return [
|
||||
f"--{flag}={value}"
|
||||
for flag, value in {**per_run_stress_flags, **DB_STRESS_PRESET}.items()
|
||||
]
|
||||
@@ -0,0 +1,179 @@
|
||||
#!/usr/bin/env python3
|
||||
# Copyright (c) Meta Platforms, Inc. and affiliates.
|
||||
# This source code is licensed under both the GPLv2 (found in the COPYING file in the root directory)
|
||||
# and the Apache 2.0 License (found in the LICENSE.Apache file in the root directory).
|
||||
|
||||
# pyre-strict
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import signal
|
||||
import subprocess
|
||||
|
||||
from runner_build import Run
|
||||
|
||||
logger: logging.Logger = logging.getLogger("runner")
|
||||
|
||||
|
||||
def _get_gdb() -> str:
|
||||
# gdb that drives db_stress. Override ICC_GDB to point at a newer gdb if the
|
||||
# telemetry's source comes out "@ ?:0" (an old gdb that cannot read this build's
|
||||
# debug info). Read on use, not at import, per Meta's lazy-import guidance.
|
||||
return os.environ.get("ICC_GDB", "/usr/bin/gdb")
|
||||
|
||||
|
||||
def _get_run_timeout_sec() -> float:
|
||||
# Hard per-run cap so one gdb+db_stress that hangs or deadlocks cannot block the
|
||||
# whole run set; single-stepping is slow, so keep it generous.
|
||||
return float(os.environ.get("ICC_RUN_TIMEOUT", "600"))
|
||||
|
||||
|
||||
# Every bucket classify() can return; runner_report iterates this so the set of outcomes
|
||||
# lives in one place.
|
||||
OUTCOMES: tuple[str, ...] = (
|
||||
"SDC",
|
||||
"CORRUPTION",
|
||||
"CRASH",
|
||||
"NO_EFFECT",
|
||||
"NO_INJECTION",
|
||||
"ERROR",
|
||||
)
|
||||
|
||||
# db_stress's read-back tags a corruption it catches with a "kind"; map each kind to our
|
||||
# result bucket (these are the only kinds db_stress emits).
|
||||
_CORRUPTION_BUCKET: dict[str, str] = {
|
||||
"lost": "SDC",
|
||||
"resurrected": "SDC",
|
||||
"wrong-value": "SDC",
|
||||
"detected-corruption": "CORRUPTION",
|
||||
}
|
||||
|
||||
|
||||
def execute(run: Run, injector_py: str, stress_cmd: str) -> None:
|
||||
try:
|
||||
_run_gdb(
|
||||
_gdb_command(injector_py, stress_cmd, run),
|
||||
os.path.join(run.dir, "gdb.log"),
|
||||
)
|
||||
except subprocess.TimeoutExpired:
|
||||
logger.warning(
|
||||
"run %d: timed out after %.0fs", run.index, _get_run_timeout_sec()
|
||||
)
|
||||
run.outcome = "ERROR"
|
||||
return
|
||||
except (
|
||||
OSError
|
||||
) as e: # gdb itself failed to launch -- a runner failure, not an outcome
|
||||
logger.warning("run %d: gdb launch failed: %s", run.index, e)
|
||||
run.outcome = "ERROR"
|
||||
return
|
||||
run.inject_json = _read_json(os.path.join(run.dir, "inject.json"))
|
||||
run.outcome = classify(run)
|
||||
|
||||
|
||||
def classify(run: Run) -> str:
|
||||
# ERROR is a runner-level failure (gdb or the injector itself), never a db_stress
|
||||
# outcome. injection_result is authoritative: only a corruption that actually landed
|
||||
# can be SDC / CORRUPTION / CRASH / NO_EFFECT.
|
||||
inject = run.inject_json
|
||||
if inject is None: # gdb died before the injector wrote inject.json
|
||||
return "ERROR"
|
||||
injection_result = inject.get("injection_result")
|
||||
if injection_result == "error":
|
||||
return "ERROR"
|
||||
if (
|
||||
injection_result != "injected"
|
||||
): # the op was never reached, nothing was corrupted
|
||||
return "NO_INJECTION"
|
||||
kind = _data_corruption_kind(run.dir)
|
||||
if kind is not None:
|
||||
return _CORRUPTION_BUCKET[kind]
|
||||
return "CRASH" if inject.get("db_stress_crash_signal") is not None else "NO_EFFECT"
|
||||
|
||||
|
||||
def _data_corruption_kind(run_dir: str) -> str | None:
|
||||
names = [
|
||||
name
|
||||
for name in os.listdir(run_dir)
|
||||
if name.startswith("data_corruption.") and name.endswith(".json")
|
||||
]
|
||||
assert (
|
||||
len(names) <= 1
|
||||
), f"expected <=1 data_corruption json in {run_dir}, got {names}"
|
||||
if not names:
|
||||
return None
|
||||
data = _read_json(os.path.join(run_dir, names[0]))
|
||||
return str(data["kind"]) if data is not None else None
|
||||
|
||||
|
||||
def _gdb_command(injector_py: str, stress_cmd: str, run: Run) -> list[str]:
|
||||
# gdb's -x runs injector.py with an empty argv, so seed its argv with -iex first;
|
||||
# --args then hands db_stress and its flags to the inferior gdb launches.
|
||||
seed_argv = "py import sys; sys.argv=" + json.dumps(_injector_argv(run))
|
||||
return [
|
||||
_get_gdb(),
|
||||
"--batch",
|
||||
"--nx",
|
||||
"-iex",
|
||||
seed_argv,
|
||||
"-x",
|
||||
injector_py,
|
||||
"--args",
|
||||
stress_cmd,
|
||||
*run.stress_flags,
|
||||
]
|
||||
|
||||
|
||||
def _injector_argv(run: Run) -> list[str]:
|
||||
return [
|
||||
"injector.py",
|
||||
"--op",
|
||||
run.op,
|
||||
"--op_index",
|
||||
str(run.op_index),
|
||||
"--entry_fn",
|
||||
run.entry_fn,
|
||||
"--target_fn",
|
||||
run.target_fn,
|
||||
"--seed",
|
||||
str(run.seed),
|
||||
"--dir",
|
||||
run.dir,
|
||||
]
|
||||
|
||||
|
||||
def _run_gdb(command: list[str], log_path: str) -> None:
|
||||
with open(log_path, "wb") as log_file:
|
||||
proc = subprocess.Popen(
|
||||
command,
|
||||
stdout=log_file,
|
||||
stderr=subprocess.STDOUT, # fold gdb's stderr into the same gdb.log
|
||||
# new session => its own process group, so a timeout kills gdb AND db_stress
|
||||
start_new_session=True,
|
||||
)
|
||||
try:
|
||||
proc.wait(timeout=_get_run_timeout_sec())
|
||||
except subprocess.TimeoutExpired:
|
||||
_kill_group(proc)
|
||||
proc.wait()
|
||||
raise
|
||||
|
||||
|
||||
def _kill_group(proc: subprocess.Popen[bytes]) -> None:
|
||||
try:
|
||||
os.killpg(os.getpgid(proc.pid), signal.SIGKILL)
|
||||
except ProcessLookupError:
|
||||
pass # already exited
|
||||
|
||||
|
||||
def _read_json(path: str) -> dict[str, object] | None:
|
||||
# None if the file is missing or truncated (e.g. a crash mid-write); the caller reads
|
||||
# a missing inject.json as ERROR.
|
||||
try:
|
||||
with open(path) as f:
|
||||
return json.load(f)
|
||||
except (FileNotFoundError, json.JSONDecodeError):
|
||||
return None
|
||||
@@ -0,0 +1,46 @@
|
||||
#!/usr/bin/env python3
|
||||
# Copyright (c) Meta Platforms, Inc. and affiliates.
|
||||
# This source code is licensed under both the GPLv2 (found in the COPYING file in the root directory)
|
||||
# and the Apache 2.0 License (found in the LICENSE.Apache file in the root directory).
|
||||
|
||||
# pyre-strict
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
from collections import Counter
|
||||
from dataclasses import asdict
|
||||
|
||||
from runner_build import Run
|
||||
from runner_execute import OUTCOMES
|
||||
|
||||
logger: logging.Logger = logging.getLogger("runner")
|
||||
|
||||
|
||||
def report(
|
||||
runs: list[Run], op: str, report_dir: str, base_seed: int
|
||||
) -> dict[str, object]:
|
||||
counts = Counter(run.outcome for run in runs)
|
||||
summary: dict[str, object] = {
|
||||
"op": op,
|
||||
"base_seed": base_seed, # re-run with --seed=<base_seed>; run i used base_seed+i
|
||||
"outcome_counts": {bucket: counts.get(bucket, 0) for bucket in OUTCOMES},
|
||||
"runs": [asdict(run) for run in runs],
|
||||
}
|
||||
with open(os.path.join(report_dir, "summary.json"), "w", encoding="utf-8") as f:
|
||||
json.dump(summary, f, indent=2)
|
||||
|
||||
errors = counts.get("ERROR", 0)
|
||||
if errors:
|
||||
# ERROR is a per-run harness failure, not a db_stress data outcome, and separate
|
||||
# from the setup failures that fail fast in the preflight. Two causes, both rare
|
||||
# but expected: a timed-out run (a deadlock from the injected corruption, or just
|
||||
# slow under load), or gdb crashing mid-run / an injector bug (rarer, and worth
|
||||
# investigating -- it may point at something to fix). Surface the count, never fail
|
||||
# the sweep (no retry by design); dig in if it climbs.
|
||||
logger.warning(
|
||||
"%d/%d runs ended in ERROR; monitor that this stays low", errors, len(runs)
|
||||
)
|
||||
return summary
|
||||
Reference in new issue
Block a user