diff --git a/src/cmd/src/bin/query_perf_fixture/case.rs b/src/cmd/src/bin/query_perf_fixture/case.rs index 03319c2c4c0..8474b52d0a7 100644 --- a/src/cmd/src/bin/query_perf_fixture/case.rs +++ b/src/cmd/src/bin/query_perf_fixture/case.rs @@ -85,6 +85,56 @@ pub(super) struct OtlpTraceLoadThresholds { pub(super) struct WriteThroughputScenario { pub(super) remote_write: PromRemoteWritePlan, pub(super) write_measure: WriteMeasureConfig, + /// Workload-scheduler configuration injected into the datanode via + /// environment variables. When absent the runner leaves both targets on + /// the datanode default (scheduler disabled). `enable` is not used by the + /// runner for derivation: the base target always runs with the scheduler + /// disabled and the candidate target always runs with it enabled when this + /// section is present. + #[serde(default)] + pub(super) scheduler: Option, +} + +#[derive(Debug, Deserialize, Serialize)] +#[serde(deny_unknown_fields)] +pub(super) struct WorkloadSchedulerCaseConfig { + /// Enables policy-controlled query and write task spawning. Mirrors the + /// datanode `runtime.experimental_workload_scheduler.enable` option; the + /// runner derives the per-target value (base = false, candidate = true) + /// and ignores this field for derivation. + #[serde(default)] + pub(super) enable: bool, + /// Maximum polls admitted to Tokio at once. Zero uses four times + /// `global_rt_size`, consistent with the runtime default. + #[serde(default)] + pub(super) max_concurrent_polls: u64, + /// Relative share for query polls while writes are also backlogged. + #[serde(default = "default_scheduler_query_weight")] + pub(super) query_weight: u64, + /// Relative share for write polls while queries are also backlogged. + #[serde(default = "default_scheduler_write_weight")] + pub(super) write_weight: u64, +} + +pub(super) fn default_scheduler_query_weight() -> u64 { + 2 +} +pub(super) fn default_scheduler_write_weight() -> u64 { + 8 +} + +impl WorkloadSchedulerCaseConfig { + pub(super) fn validate(&self) -> Result<(), String> { + for (name, value) in [ + ("query_weight", self.query_weight), + ("write_weight", self.write_weight), + ] { + if value == 0 { + return Err(format!("scenario.scheduler.{name} must be positive")); + } + } + Ok(()) + } } #[derive(Debug, Deserialize, Serialize)] @@ -647,6 +697,28 @@ window_seconds = 5 max_failure_rate = 0.05 max_mean_rps_regression_pct = 10.0 max_p99_latency_regression_pct = 10.0 +"#; + + const SCHEDULER_CASE: &str = r#" +[scenario] +kind = "write_throughput" + +[scenario.remote_write] +metric = "write_throughput_test" + +[scenario.write_measure] +duration_seconds = 60 +window_seconds = 5 + +[scenario.write_measure.thresholds] +max_failure_rate = 0.05 +max_mean_rps_regression_pct = 10.0 +max_p99_latency_regression_pct = 10.0 + +[scenario.scheduler] +max_concurrent_polls = 16 +query_weight = 2 +write_weight = 8 "#; #[test] @@ -676,6 +748,16 @@ max_p99_latency_regression_pct = 10.0 .max_p99_latency_regression_pct, 10.0 ); + let scheduler = scenario + .scheduler + .expect("built-in case must declare a scheduler section"); + assert!(!scheduler.enable); + assert_eq!(scheduler.max_concurrent_polls, 16); + assert_eq!(scheduler.query_weight, 2); + assert_eq!(scheduler.write_weight, 8); + scheduler + .validate() + .expect("built-in scheduler must validate"); scenario .write_measure .validate() @@ -688,6 +770,7 @@ max_p99_latency_regression_pct = 10.0 assert_eq!(scenario.remote_write.series_count, 8); assert_eq!(scenario.remote_write.samples_per_series, 30); assert_eq!(scenario.write_measure.target_rps, 0.0); + assert!(scenario.scheduler.is_none()); assert!(scenario.write_measure.thresholds.min_rps_absolute.is_none()); scenario .write_measure @@ -695,6 +778,52 @@ max_p99_latency_regression_pct = 10.0 .expect("minimal case must validate"); } + #[test] + fn write_throughput_parses_scheduler_section() { + let scenario = write_throughput_scenario(SCHEDULER_CASE); + let scheduler = scenario + .scheduler + .expect("scheduler section must parse into Some"); + assert!(!scheduler.enable); + assert_eq!(scheduler.max_concurrent_polls, 16); + assert_eq!(scheduler.query_weight, 2); + assert_eq!(scheduler.write_weight, 8); + scheduler + .validate() + .expect("scheduler config must validate"); + } + + #[test] + fn write_throughput_rejects_invalid_scheduler_weights() { + let zero_query = SCHEDULER_CASE.replace("query_weight = 2", "query_weight = 0"); + let scenario = write_throughput_scenario(&zero_query); + let err = scenario + .scheduler + .unwrap() + .validate() + .expect_err("zero query_weight must be rejected"); + assert!(err.contains("query_weight"), "{err}"); + + let zero_write = SCHEDULER_CASE.replace("write_weight = 8", "write_weight = 0"); + let scenario = write_throughput_scenario(&zero_write); + let err = scenario + .scheduler + .unwrap() + .validate() + .expect_err("zero write_weight must be rejected"); + assert!(err.contains("write_weight"), "{err}"); + } + + #[test] + fn write_throughput_rejects_unknown_scheduler_fields() { + let unknown = SCHEDULER_CASE.replace( + "max_concurrent_polls = 16", + "max_concurrent_polls = 16\nbogus_scheduler_field = 1", + ); + let err = parse_case(&unknown).expect_err("unknown scheduler field must be rejected"); + assert!(err.to_string().contains("bogus_scheduler_field"), "{err}"); + } + #[test] fn rejects_bad_types() { // window_seconds must be an integer, not a string. diff --git a/src/cmd/src/bin/query_perf_fixture/mod.rs b/src/cmd/src/bin/query_perf_fixture/mod.rs index 4abc656c6f2..41349b7a67d 100644 --- a/src/cmd/src/bin/query_perf_fixture/mod.rs +++ b/src/cmd/src/bin/query_perf_fixture/mod.rs @@ -147,10 +147,15 @@ fn run_plan(args: PlanArgs) -> Result<(), Box> { } } } - if let Scenario::WriteThroughput(s) = &case.scenario - && let Err(message) = s.write_measure.validate() - { - return Err(message.into()); + if let Scenario::WriteThroughput(s) = &case.scenario { + if let Err(message) = s.write_measure.validate() { + return Err(message.into()); + } + if let Some(scheduler) = &s.scheduler + && let Err(message) = scheduler.validate() + { + return Err(message.into()); + } } println!( "{}", diff --git a/tests/perf/README.md b/tests/perf/README.md index 17088f6719b..373316cafc7 100644 --- a/tests/perf/README.md +++ b/tests/perf/README.md @@ -288,6 +288,17 @@ throughput must not regress relative to base. [scenario] kind = "write_throughput" +[scenario.scheduler] +# Optional. The runner derives the enable flag per target and injects it into +# the datanode as environment variables: the base datanode always starts with +# GREPTIMEDB_DATANODE__RUNTIME__EXPERIMENTAL_WORKLOAD_SCHEDULER__ENABLE=false +# and the candidate datanode with ENABLE=true plus the values below +# (max_concurrent_polls 0 = 4 * global_rt_size). Without this section both +# targets run with the scheduler disabled. +max_concurrent_polls = 16 +query_weight = 2 +write_weight = 8 + [scenario.remote_write] database = "public" metric = "write_throughput_scheduler" @@ -328,6 +339,18 @@ positive integers, `window_seconds` must divide `duration_seconds`, `max_failure_rate` must be in `[0, 1]`. Unknown fields are rejected by `query_perf_fixture plan`. +The optional `[scenario.scheduler]` section (deny-unknown-fields) configures +the datanode workload scheduler: `enable` (default false, derived by the +runner rather than set in the case), `max_concurrent_polls` (default 0 = +4 * `global_rt_size`), and `query_weight`/`write_weight` (defaults 2 and 8) +which must be positive. The runner injects the scheduler into the datanode via +environment variables (`GREPTIMEDB_DATANODE__RUNTIME__EXPERIMENTAL_WORKLOAD_SCHEDULER__ENABLE` +etc.) because the datanode is spawned with CLI args only and no +`--config-file`; metasrv and frontend never receive scheduler env. Scheduler +on/off is the base/candidate axis of this case: base = off, candidate = on, +and each target report entry records the applied `scheduler` config under +`{enabled, max_concurrent_polls, query_weight, write_weight}`. + For each target the runner writes all sample chunks, flushes the physical table, and computes: diff --git a/tests/perf/query_cases/write_throughput_scheduler/case.toml b/tests/perf/query_cases/write_throughput_scheduler/case.toml index 1ec2ba2623d..3712d80f379 100644 --- a/tests/perf/query_cases/write_throughput_scheduler/case.toml +++ b/tests/perf/query_cases/write_throughput_scheduler/case.toml @@ -4,11 +4,18 @@ # (scheduler off) vs candidate (scheduler on) under a single-class backlog: # the scheduler must be work-conserving and must not regress write throughput. # +# The runner derives the scheduler enable flag per target from the +# [scenario.scheduler] section below: the base target always starts its +# datanode with GREPTIMEDB_DATANODE__RUNTIME__EXPERIMENTAL_WORKLOAD_SCHEDULER__ENABLE=false +# and the candidate target with ENABLE=true plus the weights/max_polls below +# (injected as datanode environment variables, since the runner spawns the +# datanode with CLI args only). `enable` is intentionally not set here. +# # Sized for a local run: 2048 series x 3600 samples = 7,372,800 rows per target. # The ingestion is split into 20 sample chunks of 180 samples per series; the # measurement spans the first 60s (12 windows of 5s). Gates: <= 5% failed # chunks, <= 10% mean-RPS regression, <= 10% p99-latency regression, and an -# absolute floor of 50k rows/s per target. +# absolute floor of 15k rows/s per target. [case] name = "write_throughput_scheduler" @@ -18,6 +25,13 @@ issue = "https://github.com/GreptimeTeam/greptimedb/pull/8736" [scenario] kind = "write_throughput" +[scenario.scheduler] +# enable is not set here: the runner derives it per target (base = disabled, +# candidate = enabled). max_concurrent_polls 0 = 4 * global_rt_size. +max_concurrent_polls = 16 +query_weight = 2 +write_weight = 8 + [scenario.remote_write] database = "public" metric = "write_throughput_scheduler" diff --git a/tests/perf/query_regression_runner.py b/tests/perf/query_regression_runner.py index 444bfc33d8c..a282862bfad 100644 --- a/tests/perf/query_regression_runner.py +++ b/tests/perf/query_regression_runner.py @@ -34,7 +34,7 @@ import tomllib import urllib.error import urllib.parse import urllib.request -from dataclasses import dataclass +from dataclasses import dataclass, field from pathlib import Path from typing import Any @@ -66,6 +66,7 @@ class RunTarget: datanode_rpc_port: int datanode_http_port: int datanode_data_dir: Path + scheduler_env: dict[str, str] = field(default_factory=dict) def parse_args() -> argparse.Namespace: @@ -126,7 +127,7 @@ def allocate_ports(n: int) -> list[int]: s.close() -def make_target(name: str, binary: Path, root: Path, ports: list[int], fixture_dir: Path | None = None) -> RunTarget: +def make_target(name: str, binary: Path, root: Path, ports: list[int], fixture_dir: Path | None = None, scheduler_env: dict[str, str] | None = None) -> RunTarget: work_dir = root / name return RunTarget( name, @@ -144,9 +145,55 @@ def make_target(name: str, binary: Path, root: Path, ports: list[int], fixture_d ports[2], ports[3], work_dir / "datanode-0" / "data", + scheduler_env or {}, ) +def scheduler_env(enable: bool, scheduler: dict[str, Any] | None) -> dict[str, str]: + """Derive the datanode workload-scheduler environment variables for a target. + + The datanode is spawned with CLI args only (no --config-file), so the + scheduler must be injected via environment. Config env vars use the + `PREFIX__KEY__SUBKEY` form (see src/common/config/src/config.rs + load_layered_options), and the datanode env prefix defaults to + `GREPTIMEDB_DATANODE`; the scheduler lives at + `runtime.experimental_workload_scheduler`. + + Returns an empty dict when the case has no `[scenario.scheduler]` section: + both base and candidate then run with the datanode default (scheduler + disabled). When the section is present the base target only pins + `ENABLE=false` while the candidate pins `ENABLE=true` and forwards the + scheduler weights/max-polls so both targets are configured identically + except for the enable flag. + """ + if scheduler is None: + return {} + prefix = "GREPTIMEDB_DATANODE__RUNTIME__EXPERIMENTAL_WORKLOAD_SCHEDULER" + env = {f"{prefix}__ENABLE": "true" if enable else "false"} + if enable: + for key in ("max_concurrent_polls", "query_weight", "write_weight"): + if key in scheduler: + env[f"{prefix}__{key.upper()}"] = str(scheduler[key]) + return env + + +def scheduler_report_entry(target_name: str, scheduler: dict[str, Any] | None) -> dict[str, Any]: + """Scheduler configuration applied to a target, for the target report entry. + + The base target always runs with the scheduler disabled; the candidate + runs with it enabled iff the case has a `[scenario.scheduler]` section. + Defaults mirror the datanode runtime defaults (max_concurrent_polls 0 = + four times global_rt_size, query_weight 2, write_weight 8). + """ + enabled = target_name == "candidate" and scheduler is not None + return { + "enabled": enabled, + "max_concurrent_polls": int(scheduler["max_concurrent_polls"]) if scheduler else 0, + "query_weight": int(scheduler["query_weight"]) if scheduler else 2, + "write_weight": int(scheduler["write_weight"]) if scheduler else 8, + } + + def run_command(cmd: list[str], cwd: Path | None = None) -> dict[str, Any]: started = time.monotonic() proc = subprocess.run(cmd, cwd=cwd, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False) @@ -340,11 +387,11 @@ class DistributedCluster: }, } - def _spawn(self, name: str, args: list[str]) -> None: + def _spawn(self, name: str, args: list[str], env: dict[str, str] | None = None) -> None: logs = self.target.work_dir / "logs" / name logs.mkdir(parents=True, exist_ok=True) with (logs / "stdout.log").open("ab") as out, (logs / "stderr.log").open("ab") as err: - self.procs[name] = subprocess.Popen(args, stdout=out, stderr=err) + self.procs[name] = subprocess.Popen(args, stdout=out, stderr=err, env=env) def _ensure_metasrv_alive(self) -> None: proc = self.procs.get("metasrv") @@ -400,6 +447,9 @@ class DistributedCluster: "--metasrv-addrs", f"127.0.0.1:{self.target.metasrv_rpc_port}", ], + env=None + if not self.target.scheduler_env + else {**os.environ, **self.target.scheduler_env}, ) wait_health(self.target.datanode_http_port) @@ -1764,6 +1814,7 @@ def run_write_throughput_scenario(args: argparse.Namespace, case: dict[str, Any] "remote_write": rw, "flushes": flushes, "flush": flushes[-1] if flushes else None, + "scheduler": scheduler_report_entry(target.name, scenario_config.get("scheduler")), "write_measurement": measurement, } flushes_ok = all(flush.get("ok") for flush in flushes) @@ -1861,7 +1912,11 @@ def main() -> int: if scenario_kind == "direct_readable_sst" and not args.dry_run and args.fixture_generator is None: raise ValueError("--fixture-generator is required unless --dry-run is set") ports = allocate_ports(16) - targets = [make_target("base", args.base_bin.resolve(), work_root, ports[:8]), make_target("candidate", args.candidate_bin.resolve(), work_root, ports[8:])] + scheduler_cfg = scenario_config.get("scheduler") + targets = [ + make_target("base", args.base_bin.resolve(), work_root, ports[:8], scheduler_env=scheduler_env(False, scheduler_cfg)), + make_target("candidate", args.candidate_bin.resolve(), work_root, ports[8:], scheduler_env=scheduler_env(True, scheduler_cfg)), + ] require_fresh_work_dirs(targets, reuse_work_dir=args.reuse_work_dir, dry_run=args.dry_run, fixture_only=args.fixture_only) fixture_dir = fixture_root(work_root, case_path, case, args.fixture_cache_dir) reuse_fixture = args.reuse_fixture or args.fixture_cache_dir is not None diff --git a/tests/perf/test_write_throughput_runner.py b/tests/perf/test_write_throughput_runner.py index 5a8da2c9247..1d7cee2cebd 100644 --- a/tests/perf/test_write_throughput_runner.py +++ b/tests/perf/test_write_throughput_runner.py @@ -36,6 +36,13 @@ sys.modules[SPEC.name] = runner SPEC.loader.exec_module(runner) +class _FakeProc: + """Minimal stand-in for a live subprocess (poll() returns None = running).""" + + def poll(self) -> None: + return None + + def make_remote(**overrides): remote = { "database": "public", @@ -230,6 +237,120 @@ class WriteThroughputThresholdTest(unittest.TestCase): self.assertEqual([p["threshold"] for p in planned], ["max_failure_rate", "max_mean_rps_regression_pct", "max_p99_latency_regression_pct"]) +class WriteThroughputSchedulerEnvTest(unittest.TestCase): + """Scheduler on/off env derivation and datanode spawn plumbing.""" + + PREFIX = "GREPTIMEDB_DATANODE__RUNTIME__EXPERIMENTAL_WORKLOAD_SCHEDULER" + SCHEDULER = {"max_concurrent_polls": 16, "query_weight": 2, "write_weight": 8} + + def test_scheduler_env_absent_section_disables_both(self) -> None: + self.assertEqual(runner.scheduler_env(False, None), {}) + self.assertEqual(runner.scheduler_env(True, None), {}) + + def test_scheduler_env_derivation_base_disables_candidate_enables(self) -> None: + base_env = runner.scheduler_env(False, self.SCHEDULER) + self.assertEqual(base_env, {f"{self.PREFIX}__ENABLE": "false"}) + candidate_env = runner.scheduler_env(True, self.SCHEDULER) + self.assertEqual( + candidate_env, + { + f"{self.PREFIX}__ENABLE": "true", + f"{self.PREFIX}__MAX_CONCURRENT_POLLS": "16", + f"{self.PREFIX}__QUERY_WEIGHT": "2", + f"{self.PREFIX}__WRITE_WEIGHT": "8", + }, + ) + + def test_make_target_populates_scheduler_env(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + base = runner.make_target( + "base", Path("/bin/true"), root, list(range(10_000, 10_008)), + scheduler_env=runner.scheduler_env(False, self.SCHEDULER), + ) + candidate = runner.make_target( + "candidate", Path("/bin/true"), root, list(range(10_008, 10_016)), + scheduler_env=runner.scheduler_env(True, self.SCHEDULER), + ) + no_scheduler = runner.make_target( + "candidate", Path("/bin/true"), root, list(range(10_016, 10_024)), + ) + self.assertEqual(base.scheduler_env, {f"{self.PREFIX}__ENABLE": "false"}) + self.assertEqual(candidate.scheduler_env[f"{self.PREFIX}__ENABLE"], "true") + self.assertEqual(candidate.scheduler_env[f"{self.PREFIX}__MAX_CONCURRENT_POLLS"], "16") + self.assertEqual(candidate.scheduler_env[f"{self.PREFIX}__QUERY_WEIGHT"], "2") + self.assertEqual(candidate.scheduler_env[f"{self.PREFIX}__WRITE_WEIGHT"], "8") + self.assertEqual(no_scheduler.scheduler_env, {}) + + def test_start_datanode_passes_scheduler_env(self) -> None: + def spawn_env(target: runner.RunTarget) -> dict[str, str] | None: + cluster = runner.DistributedCluster(target) + cluster.procs["metasrv"] = _FakeProc() + with ( + patch.object(runner.subprocess, "Popen") as popen, + patch.object(runner, "wait_health"), + ): + cluster.start_datanode() + popen.assert_called_once() + return popen.call_args.kwargs.get("env") + + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + base = runner.make_target( + "base", Path("/bin/true"), root, list(range(10_000, 10_008)), + scheduler_env=runner.scheduler_env(False, self.SCHEDULER), + ) + candidate = runner.make_target( + "candidate", Path("/bin/true"), root, list(range(10_008, 10_016)), + scheduler_env=runner.scheduler_env(True, self.SCHEDULER), + ) + base_env = spawn_env(base) + self.assertIsNotNone(base_env) + self.assertEqual(base_env[f"{self.PREFIX}__ENABLE"], "false") + candidate_env = spawn_env(candidate) + self.assertIsNotNone(candidate_env) + self.assertEqual(candidate_env[f"{self.PREFIX}__ENABLE"], "true") + self.assertEqual(candidate_env[f"{self.PREFIX}__MAX_CONCURRENT_POLLS"], "16") + self.assertEqual(candidate_env[f"{self.PREFIX}__QUERY_WEIGHT"], "2") + self.assertEqual(candidate_env[f"{self.PREFIX}__WRITE_WEIGHT"], "8") + # The scheduler env is merged over os.environ so the datanode keeps + # PATH and friends. + self.assertIn("PATH", candidate_env) + + def test_metasrv_and_frontend_do_not_receive_scheduler_env(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + candidate = runner.make_target( + "candidate", Path("/bin/true"), root, list(range(10_008, 10_016)), + scheduler_env=runner.scheduler_env(True, self.SCHEDULER), + ) + cluster = runner.DistributedCluster(candidate) + cluster.procs["metasrv"] = _FakeProc() + with ( + patch.object(runner.subprocess, "Popen") as popen, + patch.object(runner, "wait_health"), + ): + popen.return_value = _FakeProc() + cluster.start_metasrv() + cluster.start_frontend() + self.assertEqual(popen.call_count, 2) + for call in popen.call_args_list: + self.assertIsNone(call.kwargs.get("env")) + + def test_scheduler_report_entry(self) -> None: + self.assertEqual( + runner.scheduler_report_entry("base", self.SCHEDULER), + {"enabled": False, "max_concurrent_polls": 16, "query_weight": 2, "write_weight": 8}, + ) + self.assertEqual( + runner.scheduler_report_entry("candidate", self.SCHEDULER), + {"enabled": True, "max_concurrent_polls": 16, "query_weight": 2, "write_weight": 8}, + ) + disabled = {"enabled": False, "max_concurrent_polls": 0, "query_weight": 2, "write_weight": 8} + self.assertEqual(runner.scheduler_report_entry("base", None), disabled) + self.assertEqual(runner.scheduler_report_entry("candidate", None), disabled) + + class WriteThroughputScenarioDryRunTest(unittest.TestCase): def test_dry_run_plans_without_subprocess(self) -> None: events = [] @@ -277,6 +398,10 @@ class WriteThroughputScenarioDryRunTest(unittest.TestCase): self.assertEqual(report["targets"][0]["write_measurement"]["status"], "planned") self.assertEqual(len(report["thresholds"]), 4) self.assertTrue(all(t["status"] == "planned" for t in report["thresholds"])) + # No [scenario.scheduler] section: both targets run scheduler-disabled. + disabled = {"enabled": False, "max_concurrent_polls": 0, "query_weight": 2, "write_weight": 8} + self.assertEqual(report["targets"][0]["scheduler"], disabled) + self.assertEqual(report["targets"][1]["scheduler"], disabled) # The dry-run report file for each target is written under the work dir. self.assertTrue(targets[0].report_path.exists())