From f5f5d468bbc8fa68e65b46e8541a3370c68b9838 Mon Sep 17 00:00:00 2001 From: shuiyisong <113876041+shuiyisong@users.noreply.github.com> Date: Sun, 26 Jul 2026 23:08:27 +0800 Subject: [PATCH] ci: add OTLP trace ingestion regression testing (#8631) * chore: update CI config Signed-off-by: shuiyisong * chore: add CI Signed-off-by: shuiyisong * chore: update CI config Signed-off-by: shuiyisong * ci: report otelgen runner diagnostics Signed-off-by: shuiyisong * chore: add script to draw result diagram Signed-off-by: shuiyisong --------- Signed-off-by: shuiyisong --- .../query-regression/Dockerfile | 15 + .../query-regression/README.md | 18 +- .../query-regression/values-8-cores.yaml | 4 +- .github/scripts/query-regression-run.py | 3 + .github/scripts/query-regression-summary.py | 67 ++++- .github/workflows/query-regression.yml | 25 +- src/cmd/src/bin/query_perf_fixture/case.rs | 32 ++- src/cmd/src/bin/query_perf_fixture/mod.rs | 22 ++ tests/perf/README.md | 61 ++++ tests/perf/plot_otlp_trace_report.sh | 80 ++++++ .../query_cases/otlp_trace_load/case.toml | 23 ++ tests/perf/query_regression_runner.py | 263 +++++++++++++++++- ...ery_regression_runner_compaction_toctou.py | 18 ++ ...query_regression_runner_otlp_trace_load.py | 228 +++++++++++++++ .../test_query_regression_summary_otlp.py | 88 ++++++ 15 files changed, 924 insertions(+), 23 deletions(-) create mode 100755 tests/perf/plot_otlp_trace_report.sh create mode 100644 tests/perf/query_cases/otlp_trace_load/case.toml create mode 100644 tests/perf/test_query_regression_runner_otlp_trace_load.py create mode 100644 tests/perf/test_query_regression_summary_otlp.py diff --git a/.github/runner-scale-sets/query-regression/Dockerfile b/.github/runner-scale-sets/query-regression/Dockerfile index 6139291958..d3e89e8176 100644 --- a/.github/runner-scale-sets/query-regression/Dockerfile +++ b/.github/runner-scale-sets/query-regression/Dockerfile @@ -1,3 +1,17 @@ +FROM golang:1.23@sha256:e87b2a5f6df2dff71ea330d55d54f4979eb380ae58a7e3aabc9d53121243e689 AS otelgen + +ARG OTELGEN_COMMIT=863a3f395d062c7322cc1de08a38774b7fdaa6c8 + +RUN git init /tmp/otelgen \ + && git -C /tmp/otelgen remote add origin https://github.com/WenyXu/otelgen.git \ + && git -C /tmp/otelgen fetch --depth=1 origin "${OTELGEN_COMMIT}" \ + && test "$(git -C /tmp/otelgen rev-parse FETCH_HEAD)" = "${OTELGEN_COMMIT}" \ + && git -C /tmp/otelgen checkout --detach FETCH_HEAD \ + && cd /tmp/otelgen \ + && CGO_ENABLED=0 go build -trimpath \ + -ldflags "-s -w -X main.version=${OTELGEN_COMMIT} -X main.commit=${OTELGEN_COMMIT}" \ + -o /usr/local/bin/otelgen ./cmd/otelgen + FROM ghcr.io/astral-sh/uv@sha256:3d868e555f8f1dbc324afa005066cd11e1053fc4743b9808ca8025283e65efa5 AS uv FROM ghcr.io/actions/actions-runner@sha256:08c30b0a7105f64bddfc485d2487a22aa03932a791402393352fdf674bda2c29 @@ -43,6 +57,7 @@ RUN apt-get update \ && rm -rf /var/lib/apt/lists/* COPY --from=uv /uv /uvx /usr/local/bin/ +COPY --from=otelgen /usr/local/bin/otelgen /usr/local/bin/otelgen RUN curl --fail --location --silent --show-error \ --output /tmp/sccache.tar.gz \ diff --git a/.github/runner-scale-sets/query-regression/README.md b/.github/runner-scale-sets/query-regression/README.md index 4fe8c0f3af..307d42396f 100644 --- a/.github/runner-scale-sets/query-regression/README.md +++ b/.github/runner-scale-sets/query-regression/README.md @@ -68,10 +68,15 @@ operator before any canary. ## Runner image and workflow tools Build and push the derived runner image; it preserves the official -`/home/runner/run.sh` entrypoint and supplies CI tools needed at runtime: +`/home/runner/run.sh` entrypoint and supplies CI tools needed at runtime. The +image builds `otelgen` from +[`WenyXu/otelgen`](https://github.com/WenyXu/otelgen) commit +[`863a3f395d062c7322cc1de08a38774b7fdaa6c8`](https://github.com/WenyXu/otelgen/commit/863a3f395d062c7322cc1de08a38774b7fdaa6c8) +so trace cases do not download or compile tools during a benchmark run: ```bash docker build \ + --platform linux/amd64 \ -f .github/runner-scale-sets/query-regression/Dockerfile \ -t greptime-registry.cn-hangzhou.cr.aliyuncs.com/greptime/greptimedb-query-regression-runner:latest \ .github/runner-scale-sets/query-regression @@ -79,9 +84,11 @@ docker build \ docker push greptime-registry.cn-hangzhou.cr.aliyuncs.com/greptime/greptimedb-query-regression-runner:latest ``` -Deploy by digest, not mutable tag, by updating `values-8-cores.yaml` after a -rebuild. If the registry is private, use a dedicated read-only pull secret only -as `imagePullSecrets`; never expose registry credentials to runner containers. +Deploy by digest, not mutable tag, by updating both image references in +`values-8-cores.yaml` after a rebuild. Update `RUNNER_IMAGE_DIGEST` and bump +`RUNNER_IMAGE_EPOCH` in `query-regression.yml` at the same time. If the registry +is private, use a dedicated read-only pull secret only as `imagePullSecrets`; +never expose registry credentials to runner containers. Both digest-pinned init and runner containers use `IfNotPresent`: the immutable digest makes a cached image safe and avoids adding a registry dependency to every runner startup. @@ -94,7 +101,8 @@ credentials or secrets must never be placed in a ConfigMap. Before builds, the workflow asserts UID/GID 1001 and exact image tool versions: `libprotoc 3.21.12`, `uv 0.11.26`, `mold 2.30.0`, `Python 3.12.3`, `sccache -0.16.0`, root-owned `rustup 1.29.0`, and the image-baked +0.16.0`, `otelgen` commit `863a3f395d062c7322cc1de08a38774b7fdaa6c8`, +root-owned `rustup 1.29.0`, and the image-baked `nightly-2026-03-21` Rust toolchain. Rustup, Cargo, and Rustc must resolve from `/opt/cargo/bin`; the runner cannot write `/opt/rustup` or `/opt/cargo/bin`. Protobuf well-known includes, including `google/protobuf/any.proto` and diff --git a/.github/runner-scale-sets/query-regression/values-8-cores.yaml b/.github/runner-scale-sets/query-regression/values-8-cores.yaml index 057ff55b50..cff1322680 100644 --- a/.github/runner-scale-sets/query-regression/values-8-cores.yaml +++ b/.github/runner-scale-sets/query-regression/values-8-cores.yaml @@ -27,7 +27,7 @@ template: claimName: query-regression-build-cache initContainers: - name: initialize-build-cache - image: greptime-registry.cn-hangzhou.cr.aliyuncs.com/greptime/greptimedb-query-regression-runner@sha256:2436278bf0756700c718927907cdbedac64d55c593f80b515acb44dec8b02cc1 + image: greptime-registry.cn-hangzhou.cr.aliyuncs.com/greptime/greptimedb-query-regression-runner@sha256:e713b294e23b7e15184e558866c90025e59930033e72c97650dbc7f1ca022d11 imagePullPolicy: IfNotPresent command: - /bin/sh @@ -65,7 +65,7 @@ template: mountPath: /cache containers: - name: runner - image: greptime-registry.cn-hangzhou.cr.aliyuncs.com/greptime/greptimedb-query-regression-runner@sha256:2436278bf0756700c718927907cdbedac64d55c593f80b515acb44dec8b02cc1 + image: greptime-registry.cn-hangzhou.cr.aliyuncs.com/greptime/greptimedb-query-regression-runner@sha256:e713b294e23b7e15184e558866c90025e59930033e72c97650dbc7f1ca022d11 imagePullPolicy: IfNotPresent command: ["/home/runner/run.sh"] env: diff --git a/.github/scripts/query-regression-run.py b/.github/scripts/query-regression-run.py index 1aa8482537..aa30aab4fb 100644 --- a/.github/scripts/query-regression-run.py +++ b/.github/scripts/query-regression-run.py @@ -118,6 +118,8 @@ def run_case(args: argparse.Namespace, case_path: Path, work_dir: Path) -> int: ] if parse_bool(args.allow_large_fixture): cmd.append("--allow-large-fixture") + if args.otelgen_bin is not None: + cmd.extend(["--otelgen-bin", str(args.otelgen_bin)]) print(f"::group::Query regression case: {case_path}", flush=True) try: @@ -159,6 +161,7 @@ def main() -> int: type=Path, default=configured_path(os.environ.get("FIXTURE_GENERATOR")), ) + parser.add_argument("--otelgen-bin", type=Path, default=configured_path(os.environ.get("OTELGEN_BIN"))) parser.add_argument("--cargo-profile", default=os.environ.get("CARGO_PROFILE", "nightly")) parser.add_argument("--work-dir", default=Path("query-regression-work"), type=Path) parser.add_argument("--http-timeout", default=os.environ.get("HTTP_TIMEOUT", "300")) diff --git a/.github/scripts/query-regression-summary.py b/.github/scripts/query-regression-summary.py index f07c7d99a1..531c513ea2 100644 --- a/.github/scripts/query-regression-summary.py +++ b/.github/scripts/query-regression-summary.py @@ -32,15 +32,25 @@ def fmt_ms(value: Any) -> str: return "N/A" +def fmt_pct(value: Any) -> str: + formatted = fmt_ms(value) + return formatted if formatted == "N/A" else f"{formatted}%" + + def esc(value: Any) -> str: text = "N/A" if value is None else str(value) return text.replace("|", "\\|").replace("\n", " ") def status_emoji(status: str | None) -> str: - return {"ok": "โœ…", "measured": "โœ…", "failed": "โŒ", "planned": "๐Ÿ“", "fixture-ready": "๐Ÿงช"}.get( - status or "", "โš ๏ธ" - ) + return { + "ok": "โœ…", + "passed": "โœ…", + "measured": "โœ…", + "failed": "โŒ", + "planned": "๐Ÿ“", + "fixture-ready": "๐Ÿงช", + }.get(status or "", "โš ๏ธ") def measurement_map(target: dict[str, Any]) -> dict[str, dict[str, Any]]: @@ -114,6 +124,50 @@ def comparison_table(targets: list[dict[str, Any]], thresholds: list[dict[str, A return "\n".join(rows) +def otlp_comparison_table(targets: list[dict[str, Any]], thresholds: list[dict[str, Any]]) -> str: + rows = [ + "| Target | Accepted spans | Table rows | Spans/s | Mean HTTP latency ms | Failures |", + "| --- | ---: | ---: | ---: | ---: | ---: |", + ] + for target in targets: + metrics = target.get("metrics") or {} + visibility = target.get("visibility") or {} + rows.append( + "| {target} | {accepted} | {stored} | {rate} | {latency} | {failures} |".format( + target=esc(target.get("name")), + accepted=esc(metrics.get("accepted_spans")), + stored=esc(visibility.get("observed_rows")), + rate=fmt_ms(metrics.get("accepted_spans_per_second")), + latency=fmt_ms(metrics.get("mean_http_latency_ms")), + failures=esc(metrics.get("failure_count")), + ) + ) + + rows.extend( + [ + "", + "| Threshold | Scope | Actual | Limit | Status | Reason |", + "| --- | --- | ---: | ---: | --- | --- |", + ] + ) + for threshold in thresholds: + is_pct = "actual_pct" in threshold or "limit_pct" in threshold + actual = fmt_pct(threshold.get("actual_pct")) if is_pct else esc(threshold.get("actual")) + limit = fmt_pct(threshold.get("limit_pct")) if is_pct else esc(threshold.get("limit")) + rows.append( + "| {threshold} | {scope} | {actual} | {limit} | {status} {raw} | {reason} |".format( + threshold=esc(threshold.get("threshold")), + scope=esc(threshold.get("target") or "base vs candidate"), + actual=actual, + limit=limit, + status=status_emoji(threshold.get("status")), + raw=esc(threshold.get("status")), + reason=esc(threshold.get("reason")), + ) + ) + return "\n".join(rows) + + def build_markdown( report: dict[str, Any], report_path: Path, @@ -141,7 +195,12 @@ def build_markdown( targets = report.get("targets") or [] lines.extend(["", "### Targets", "", target_table(targets)]) - lines.extend(["", "### Query comparison", "", comparison_table(targets, report.get("thresholds") or [])]) + if (report.get("scenario") or {}).get("kind") == "otlp_trace_load": + lines.extend( + ["", "### OTLP trace comparison", "", otlp_comparison_table(targets, report.get("thresholds") or [])] + ) + else: + lines.extend(["", "### Query comparison", "", comparison_table(targets, report.get("thresholds") or [])]) not_enforced = [t for t in report.get("thresholds") or [] if t.get("status") == "not_enforced"] if not_enforced: diff --git a/.github/workflows/query-regression.yml b/.github/workflows/query-regression.yml index df2437a7d4..63387bc310 100644 --- a/.github/workflows/query-regression.yml +++ b/.github/workflows/query-regression.yml @@ -286,6 +286,15 @@ jobs: mold_version="$(mold --version)" [[ "${mold_version}" =~ ^mold[[:space:]]2\.30\.0([[:space:]]|$) ]] [[ "$(python3 --version)" == "Python 3.12.3" ]] + otelgen_path="$(command -v otelgen || true)" + otelgen_version="" + if [[ -n "${otelgen_path}" ]]; then + otelgen_version="$("${otelgen_path}" --version 2>&1 || true)" + fi + printf 'otelgen path: %s\n' "${otelgen_path:-}" + printf 'otelgen version: %s\n' "${otelgen_version}" + [[ "${otelgen_path}" == "/usr/local/bin/otelgen" ]] + [[ "${otelgen_version}" == *"863a3f395d062c7322cc1de08a38774b7fdaa6c8"* ]] sccache_version="$(sccache --version)" [[ "${sccache_version}" =~ ^sccache[[:space:]]0\.16\.0([[:space:]]|$) ]] [[ "$(command -v rustup)" == "/opt/cargo/bin/rustup" ]] @@ -324,8 +333,8 @@ jobs: readonly EXPECTED_CACHE_META="/home/runner/query-regression-cache-meta" readonly EXPECTED_SCCACHE_DIR="/home/runner/.cache/sccache" readonly EXPECTED_RUSTC_WRAPPER="/usr/local/bin/sccache" - readonly RUNNER_IMAGE_DIGEST="sha256:2436278bf0756700c718927907cdbedac64d55c593f80b515acb44dec8b02cc1" - readonly RUNNER_IMAGE_EPOCH="4" + readonly RUNNER_IMAGE_DIGEST="sha256:e713b294e23b7e15184e558866c90025e59930033e72c97650dbc7f1ca022d11" + readonly RUNNER_IMAGE_EPOCH="5" require_expected_root() { local name="$1" @@ -558,9 +567,12 @@ jobs: git reset --hard "${VERIFIED_CANDIDATE_SHA}" git clean -ffdx - - name: Test query regression runner lifecycle + - name: Test query regression tooling working-directory: src - run: uv run --no-project python tests/perf/test_query_regression_runner_compaction_toctou.py + run: | + uv run --no-project python tests/perf/test_query_regression_runner_compaction_toctou.py + uv run --no-project python tests/perf/test_query_regression_runner_otlp_trace_load.py + uv run --no-project python tests/perf/test_query_regression_summary_otlp.py - name: Build candidate greptime and fixture generators working-directory: src @@ -594,11 +606,13 @@ jobs: BASE_BIN: ${{ github.workspace }}/query-regression-bins/base/greptime CANDIDATE_BIN: ${{ github.workspace }}/query-regression-bins/candidate/greptime FIXTURE_GENERATOR: ${{ github.workspace }}/query-regression-bins/candidate/query_perf_fixture + OTELGEN_BIN: /usr/local/bin/otelgen + SUMMARY_SCRIPT: ${{ github.event_name == 'pull_request' && 'query-regression-trusted-scripts/query-regression-summary.py' || 'src/.github/scripts/query-regression-summary.py' }} run: >- uv run --no-project python src/.github/scripts/query-regression-run.py --base-src src --candidate-src src - --summary-script query-regression-trusted-scripts/query-regression-summary.py + --summary-script "${SUMMARY_SCRIPT}" - name: Write PR metadata for trusted comment workflow if: ${{ always() && github.event_name == 'pull_request' }} @@ -622,6 +636,7 @@ jobs: query-regression-work/**/base/report.json query-regression-work/**/candidate/report.json query-regression-work/**/logs/** + query-regression-work/**/otelgen/** query-regression-summary.md if-no-files-found: warn retention-days: 7 diff --git a/src/cmd/src/bin/query_perf_fixture/case.rs b/src/cmd/src/bin/query_perf_fixture/case.rs index 5bde9c83bf..a76c49d43d 100644 --- a/src/cmd/src/bin/query_perf_fixture/case.rs +++ b/src/cmd/src/bin/query_perf_fixture/case.rs @@ -13,7 +13,7 @@ // limitations under the License. use std::collections::HashMap; -use std::num::NonZeroUsize; +use std::num::{NonZeroU64, NonZeroUsize}; use clap::ValueEnum; use serde::{Deserialize, Serialize}; @@ -30,6 +30,8 @@ pub(super) enum Scenario { DirectReadableSst(DirectReadableSstScenario), #[serde(rename = "prom_remote_write_then_query")] PromRemoteWriteThenQuery(PromRemoteWriteThenQueryScenario), + #[serde(rename = "otlp_trace_load")] + OtlpTraceLoad(OtlpTraceLoadScenario), } #[derive(Debug, Deserialize, Serialize)] @@ -49,6 +51,33 @@ pub(super) struct PromRemoteWriteThenQueryScenario { pub(super) remote_write: PromRemoteWritePlan, } +#[derive(Debug, Deserialize, Serialize)] +pub(super) struct OtlpTraceLoadScenario { + pub(super) load: OtlpTraceLoadPlan, +} + +#[derive(Debug, Deserialize, Serialize)] +pub(super) struct OtlpTraceLoadPlan { + pub(super) database: String, + pub(super) table: String, + pub(super) pipeline: String, + pub(super) duration_seconds: NonZeroU64, + pub(super) warmup_seconds: u64, + pub(super) rate: NonZeroU64, + pub(super) workers: NonZeroUsize, + pub(super) exporter_shards: NonZeroUsize, + pub(super) workload: String, + pub(super) visibility_timeout_seconds: NonZeroU64, + pub(super) thresholds: OtlpTraceLoadThresholds, +} + +#[derive(Debug, Deserialize, Serialize)] +pub(super) struct OtlpTraceLoadThresholds { + pub(super) max_candidate_throughput_regression_pct: f64, + pub(super) max_candidate_mean_latency_regression_pct: f64, + pub(super) max_failure_count: u64, +} + #[derive(Debug, Deserialize, Serialize)] pub(super) struct PromRemoteWritePlan { #[serde(default = "default_database")] @@ -412,6 +441,7 @@ impl Scenario { match self { Scenario::DirectReadableSst(_) => "direct_readable_sst", Scenario::PromRemoteWriteThenQuery(_) => "prom_remote_write_then_query", + Scenario::OtlpTraceLoad(_) => "otlp_trace_load", } } diff --git a/src/cmd/src/bin/query_perf_fixture/mod.rs b/src/cmd/src/bin/query_perf_fixture/mod.rs index 00cef327c7..1a794c0c3b 100644 --- a/src/cmd/src/bin/query_perf_fixture/mod.rs +++ b/src/cmd/src/bin/query_perf_fixture/mod.rs @@ -125,6 +125,28 @@ fn run_plan(args: PlanArgs) -> Result<(), Box> { return Err("scenario.remote_write.read_bench requires scenario.remote_write.storage.inspect = true".into()); } } + if let Scenario::OtlpTraceLoad(s) = &case.scenario { + if s.load.warmup_seconds >= s.load.duration_seconds.get() { + return Err("scenario.load.warmup_seconds must be less than duration_seconds".into()); + } + for (name, value) in [ + ( + "max_candidate_throughput_regression_pct", + s.load.thresholds.max_candidate_throughput_regression_pct, + ), + ( + "max_candidate_mean_latency_regression_pct", + s.load.thresholds.max_candidate_mean_latency_regression_pct, + ), + ] { + if !value.is_finite() || value < 0.0 { + return Err(format!( + "scenario.load.thresholds.{name} must be a finite non-negative number" + ) + .into()); + } + } + } println!( "{}", serde_json::to_string_pretty(&json!({"schema_version": 1, "scenario": case.scenario}))? diff --git a/tests/perf/README.md b/tests/perf/README.md index 91bc6c1637..ce994b6f52 100644 --- a/tests/perf/README.md +++ b/tests/perf/README.md @@ -208,6 +208,67 @@ in 1440-sample daily time chunks, flushing after each chunk before running 1d/7d TQL selectors. It is not included in the default case set because ingestion cost dominates routine CI validation. +## OTLP trace load scenario + +`scenario.kind = "otlp_trace_load"` runs a bounded native `otelgen` process +against each local distributed cluster. The runner excludes the configured +warmup window, derives throughput and mean request latency from GreptimeDB's +OTLP-specific metrics, flushes the trace table, and verifies its row count +against the accepted-span counter. The case is intentionally outside the +default set until its variance is known. + +Build base/candidate `greptime` binaries with the same profile, build the +candidate `query_perf_fixture`, then run the case explicitly: + +```bash +WORK_DIR="$(mktemp -d /tmp/query-perf-otlp.XXXXXX)" +REPORT="$WORK_DIR/trace-report.json" +python3 tests/perf/query_regression_runner.py \ + --case tests/perf/query_cases/otlp_trace_load/case.toml \ + --base-bin /path/to/base/target/nightly/greptime \ + --candidate-bin /path/to/candidate/target/nightly/greptime \ + --fixture-generator /path/to/candidate/target/nightly/query_perf_fixture \ + --otelgen-bin /path/to/otelgen \ + --work-dir "$WORK_DIR" \ + --output "$REPORT" +``` + +Install [YouPlot](https://github.com/red-data-tools/YouPlot) once, then render +all comparison metrics and threshold results in the terminal: + +```bash +brew install youplot +tests/perf/plot_otlp_trace_report.sh "$REPORT" +``` + +For each target, `accepted_spans` should equal `table_rows`, and `failures` +should stay within `max_failure_count`. Throughput is better when +`spans_per_second` is higher; its `actual_pct` is +`(base - candidate) / base * 100`. Latency is better when `mean_latency_ms` is +lower; its `actual_pct` is `(candidate - base) / base * 100`. A positive +`actual_pct` is a candidate regression, while a negative value is an +improvement. The case passes when every `actual_pct` is at or below its +`limit_pct` and every failure-count threshold passes. For local results, run +the case at least three times on an otherwise idle machine and compare the +median regressions rather than relying on one run. + +The CI runner image includes the pinned `otelgen` binary. Until this case is +added to the default set, run it explicitly with `workflow_dispatch`: + +```bash +gh workflow run query-regression.yml \ + --ref \ + -f case=tests/perf/query_cases/otlp_trace_load/case.toml \ + -f base_ref= \ + -f candidate_ref= \ + -f cargo_profile=nightly \ + -f http_timeout=300 \ + -f runner=perf-regression-8-cores +``` + +The selected ARC scale set must already be deployed with the runner-image +digest built from the current query-regression Dockerfile. + ## Generator contract The direct-SST generator should accept a case definition with: diff --git a/tests/perf/plot_otlp_trace_report.sh b/tests/perf/plot_otlp_trace_report.sh new file mode 100755 index 0000000000..fa1e5ed796 --- /dev/null +++ b/tests/perf/plot_otlp_trace_report.sh @@ -0,0 +1,80 @@ +#!/usr/bin/env bash + +set -euo pipefail + +if [[ $# -ne 1 ]]; then + printf 'Usage: %s REPORT.json\n' "${0##*/}" >&2 + exit 2 +fi + +report="$1" + +[[ -f "${report}" ]] || { + printf 'Report does not exist: %s\n' "${report}" >&2 + exit 1 +} +command -v jq >/dev/null || { + printf 'jq is required\n' >&2 + exit 1 +} +command -v uplot >/dev/null || { + printf 'YouPlot is required; install it with: brew install youplot\n' >&2 + exit 1 +} + +jq -e ' + (.status | type == "string") and + (.targets | type == "array" and length > 0) and + all(.targets[]; + (.name | type == "string") and + (.metrics.accepted_spans | type == "number") and + (.visibility.observed_rows | type == "number") and + (.metrics.accepted_spans_per_second | type == "number") and + (.metrics.mean_http_latency_ms | type == "number") and + (.metrics.failure_count | type == "number") + ) +' "${report}" >/dev/null || { + printf 'Report is not a completed OTLP trace comparison: %s\n' "${report}" >&2 + exit 1 +} + +plot_metric() { + local title="$1" + local filter="$2" + local color="$3" + + printf '\n' + jq -r ".targets[] | [.name, ${filter}] | @tsv" "${report}" | + uplot bar -t "${title}" -w 80 -c "${color}" -o - +} + +jq -r '"Status: \(.status)"' "${report}" +row_count_status=0 +jq -r ' + "\nAccepted spans vs visible table rows", + (.targets[] | + " [\(if .metrics.accepted_spans == .visibility.observed_rows then "passed" else "failed" end)] \(.name): accepted=\(.metrics.accepted_spans), visible=\(.visibility.observed_rows)" + ) +' "${report}" +jq -e 'all(.targets[]; .metrics.accepted_spans == .visibility.observed_rows)' "${report}" >/dev/null || row_count_status=1 + +plot_metric "Accepted spans" ".metrics.accepted_spans" blue +plot_metric "Visible table rows" ".visibility.observed_rows" green +plot_metric "Throughput (spans/s, higher is better)" ".metrics.accepted_spans_per_second" cyan +plot_metric "Mean HTTP latency (ms, lower is better)" ".metrics.mean_http_latency_ms" yellow +plot_metric "Failures (lower is better)" ".metrics.failure_count" red + +jq -r ' + "\nThresholds", + (.thresholds[] | + if has("actual_pct") then + " [\(.status)] \(.threshold): \(.actual_pct)% (limit \(.limit_pct)%; base \(.base), candidate \(.candidate))" + elif has("actual") then + " [\(.status)] \(.threshold) [\(.target)]: \(.actual) (limit \(.limit))" + else + " [\(.status)] \(.threshold): \(.reason // "no measured value")" + end + ) +' "${report}" + +exit "${row_count_status}" diff --git a/tests/perf/query_cases/otlp_trace_load/case.toml b/tests/perf/query_cases/otlp_trace_load/case.toml new file mode 100644 index 0000000000..ea0e661845 --- /dev/null +++ b/tests/perf/query_cases/otlp_trace_load/case.toml @@ -0,0 +1,23 @@ +[case] +name = "otlp_trace_load" +description = "OTLP trace v1 ingestion performance regression" + +[scenario] +kind = "otlp_trace_load" + +[scenario.load] +database = "public" +table = "opentelemetry_traces" +pipeline = "greptime_trace_v1" +duration_seconds = 120 +warmup_seconds = 60 +rate = 50000 +workers = 4 +exporter_shards = 4 +workload = "microservices" +visibility_timeout_seconds = 120 + +[scenario.load.thresholds] +max_candidate_throughput_regression_pct = 20 +max_candidate_mean_latency_regression_pct = 20 +max_failure_count = 0 diff --git a/tests/perf/query_regression_runner.py b/tests/perf/query_regression_runner.py index f5c9628b83..724c38ebf1 100644 --- a/tests/perf/query_regression_runner.py +++ b/tests/perf/query_regression_runner.py @@ -21,6 +21,7 @@ import argparse import fcntl import hashlib import json +import math import os import re import shutil @@ -39,6 +40,13 @@ from typing import Any FICLONE = 0x40049409 +OTLP_TRACE_METRICS = { + "greptime_frontend_otlp_traces_rows", + "greptime_frontend_otlp_traces_failure_count", + "greptime_servers_http_otlp_traces_elapsed_sum", + "greptime_servers_http_otlp_traces_elapsed_count", +} +PROMETHEUS_SAMPLE_RE = re.compile(r"^([A-Za-z_:][A-Za-z0-9_:]*)(?:\{.*\})?\s+([^\s]+)(?:\s+.*)?$") @dataclass(frozen=True) @@ -66,9 +74,11 @@ def parse_args() -> argparse.Namespace: p.add_argument("--base-bin", required=True, type=Path) p.add_argument("--candidate-bin", required=True, type=Path) p.add_argument("--fixture-generator", type=Path) + p.add_argument("--otelgen-bin", type=Path) p.add_argument("--remote-write-generator", type=Path, help="deprecated: use --fixture-generator query_perf_fixture") p.add_argument("--storage-inspector", type=Path, help="deprecated: use --fixture-generator query_perf_fixture") p.add_argument("--work-dir", required=True, type=Path) + p.add_argument("--output", type=Path, help="write the final JSON report to this file instead of stdout") p.add_argument("--fixture-cache-dir", type=Path, help="persistent directory for generated fixtures, keyed by case content") p.add_argument("--reuse-fixture", action="store_true") p.add_argument("--allow-large-fixture", action="store_true") @@ -158,15 +168,19 @@ def column_sql(col: dict[str, Any]) -> str: def scenario(case: dict[str, Any]) -> dict[str, Any]: value = case.get("scenario") if not isinstance(value, dict): - raise ValueError("case requires [scenario] with kind = 'direct_readable_sst' or 'prom_remote_write_then_query'") + raise ValueError("case requires [scenario] with a supported kind") kind = value.get("kind") - if kind not in ("direct_readable_sst", "prom_remote_write_then_query"): - raise ValueError(f"unsupported scenario kind {kind!r}; supported: 'direct_readable_sst', 'prom_remote_write_then_query'") + supported = ("direct_readable_sst", "prom_remote_write_then_query", "otlp_trace_load") + if kind not in supported: + raise ValueError(f"unsupported scenario kind {kind!r}; supported: {', '.join(supported)}") return value def case_tables(case: dict[str, Any]) -> list[dict[str, Any]]: value = scenario(case) + if value.get("kind") == "otlp_trace_load": + load = value["load"] + return [{"database": load["database"], "name": load["table"], "engine": "trace", "validate_show_create_engine": False}] if value.get("kind") == "prom_remote_write_then_query": remote = value["remote_write"] metric = remote["metric"] @@ -1028,6 +1042,171 @@ def write_json(path: Path, payload: dict[str, Any]) -> None: path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n") +def output_report(report: dict[str, Any], output: Path | None) -> None: + if output is None: + print(json.dumps(report, indent=2, sort_keys=True)) + else: + write_json(output, report) + + +def parse_prometheus_metrics(text: str, names: set[str] = OTLP_TRACE_METRICS) -> dict[str, float]: + values: dict[str, float] = {} + for line in text.splitlines(): + match = PROMETHEUS_SAMPLE_RE.match(line.strip()) + if not match or match.group(1) not in names: + continue + value = float(match.group(2)) + if not math.isfinite(value): + raise ValueError(f"non-finite Prometheus sample for {match.group(1)}") + values[match.group(1)] = values.get(match.group(1), 0.0) + value + return values + + +def fetch_otlp_metrics(target: RunTarget, http_timeout: float) -> dict[str, Any]: + with urllib.request.urlopen(f"http://127.0.0.1:{target.http_port}/metrics", timeout=http_timeout) as response: + text = response.read().decode() + return {"captured_monotonic_seconds": time.monotonic(), "values": parse_prometheus_metrics(text)} + + +def metric_delta(after: dict[str, Any], before: dict[str, Any], name: str) -> float: + delta = float(after["values"].get(name, 0.0)) - float(before["values"].get(name, 0.0)) + if delta < 0: + raise RuntimeError(f"metric {name} decreased by {-delta}") + return delta + + +def otelgen_command(otelgen_bin: Path, target: RunTarget, load: dict[str, Any]) -> list[str]: + return [ + str(otelgen_bin), + "--protocol", "http", + "--otel-exporter-otlp-endpoint", f"127.0.0.1:{target.http_port}", + "--otel-exporter-otlp-url-path", "/v1/otlp/v1/traces", + "--header", f"x-greptime-pipeline-name={load['pipeline']}", + "--header", f"x-greptime-db-name={load['database']}", + "--header", f"x-greptime-trace-table-name={load['table']}", + "--log-level", "error", + "--insecure", + "--duration", str(int(load["duration_seconds"])), + "--rate", str(int(load["rate"])), + "traces", "multi", + "--workers", str(int(load["workers"])), + "--scenarios", str(load["workload"]), + "--exporter-shards", str(int(load["exporter_shards"])), + ] + + +def run_otelgen_load(otelgen_bin: Path | None, target: RunTarget, load: dict[str, Any], http_timeout: float, *, dry_run: bool) -> dict[str, Any]: + binary = otelgen_bin or Path("otelgen") + cmd = otelgen_command(binary, target, load) + if dry_run: + return {"status": "dry-run", "cmd": cmd} + + initial = fetch_otlp_metrics(target, http_timeout) + log_dir = target.work_dir / "otelgen" + log_dir.mkdir(parents=True, exist_ok=True) + stdout_path = log_dir / "stdout.log" + stderr_path = log_dir / "stderr.log" + duration = int(load["duration_seconds"]) + warmup = int(load["warmup_seconds"]) + timed_out = False + started = time.monotonic() + with stdout_path.open("wb") as stdout, stderr_path.open("wb") as stderr: + proc = subprocess.Popen(cmd, stdout=stdout, stderr=stderr) + try: + if warmup: + try: + proc.wait(timeout=warmup) + except subprocess.TimeoutExpired: + pass + warmed = fetch_otlp_metrics(target, http_timeout) + if proc.poll() is None: + try: + proc.wait(timeout=max(60, duration - warmup + 60)) + except subprocess.TimeoutExpired: + timed_out = True + proc.kill() + proc.wait(timeout=20) + final = fetch_otlp_metrics(target, http_timeout) + finally: + if proc.poll() is None: + proc.kill() + proc.wait(timeout=20) + elapsed = time.monotonic() - started + return { + "status": "ok" if proc.returncode == 0 and not timed_out and elapsed + 1 >= duration else "failed", + "cmd": cmd, + "returncode": proc.returncode, + "timed_out": timed_out, + "elapsed_seconds": elapsed, + "stdout_path": str(stdout_path), + "stderr_path": str(stderr_path), + "snapshots": {"initial": initial, "warmup": warmed, "final": final}, + } + + +def summarize_otlp_metrics(run: dict[str, Any]) -> dict[str, Any]: + initial = run["snapshots"]["initial"] + warmed = run["snapshots"]["warmup"] + final = run["snapshots"]["final"] + rows = "greptime_frontend_otlp_traces_rows" + failures = "greptime_frontend_otlp_traces_failure_count" + elapsed_sum = "greptime_servers_http_otlp_traces_elapsed_sum" + elapsed_count = "greptime_servers_http_otlp_traces_elapsed_count" + missing = sorted(name for name in (rows, elapsed_sum, elapsed_count) if name not in final["values"]) + accepted_spans = int(round(metric_delta(final, initial, rows))) + measurement_accepted_spans = int(round(metric_delta(final, warmed, rows))) + http_requests = int(round(metric_delta(final, warmed, elapsed_count))) + latency_seconds = metric_delta(final, warmed, elapsed_sum) + measurement_seconds = final["captured_monotonic_seconds"] - warmed["captured_monotonic_seconds"] + return { + "accepted_spans": accepted_spans, + "measurement_accepted_spans": measurement_accepted_spans, + "accepted_spans_per_second": measurement_accepted_spans / measurement_seconds if measurement_seconds > 0 else None, + "http_requests": http_requests, + "mean_http_latency_ms": latency_seconds / http_requests * 1000.0 if http_requests else None, + "failure_count": int(round(metric_delta(final, initial, failures))), + "measurement_seconds": measurement_seconds, + "missing_metrics": missing, + } + + +def planned_otlp_thresholds(load: dict[str, Any]) -> list[dict[str, Any]]: + thresholds = load["thresholds"] + return [ + {"threshold": "max_candidate_throughput_regression_pct", "status": "planned", "limit_pct": thresholds["max_candidate_throughput_regression_pct"]}, + {"threshold": "max_candidate_mean_latency_regression_pct", "status": "planned", "limit_pct": thresholds["max_candidate_mean_latency_regression_pct"]}, + {"target": "each", "threshold": "max_failure_count", "status": "planned", "limit": thresholds["max_failure_count"]}, + ] + + +def enforce_otlp_thresholds(load: dict[str, Any], base: dict[str, Any], candidate: dict[str, Any]) -> list[dict[str, Any]]: + thresholds = load["thresholds"] + results: list[dict[str, Any]] = [] + for target_name, metrics in (("base", base), ("candidate", candidate)): + failures = metrics.get("failure_count") + limit = int(thresholds["max_failure_count"]) + results.append({"target": target_name, "threshold": "max_failure_count", "status": "passed" if failures is not None and failures <= limit else "failed", "actual": failures, "limit": limit}) + + base_rate = base.get("accepted_spans_per_second") + candidate_rate = candidate.get("accepted_spans_per_second") + throughput_limit = float(thresholds["max_candidate_throughput_regression_pct"]) + if base_rate in (None, 0) or candidate_rate is None: + results.append({"threshold": "max_candidate_throughput_regression_pct", "status": "failed", "reason": "missing or zero throughput", "base": base_rate, "candidate": candidate_rate}) + else: + actual = (base_rate - candidate_rate) / base_rate * 100.0 + results.append({"threshold": "max_candidate_throughput_regression_pct", "status": "passed" if actual <= throughput_limit else "failed", "actual_pct": actual, "limit_pct": throughput_limit, "base": base_rate, "candidate": candidate_rate}) + + base_latency = base.get("mean_http_latency_ms") + candidate_latency = candidate.get("mean_http_latency_ms") + latency_limit = float(thresholds["max_candidate_mean_latency_regression_pct"]) + if base_latency in (None, 0) or candidate_latency is None: + results.append({"threshold": "max_candidate_mean_latency_regression_pct", "status": "failed", "reason": "missing or zero mean latency", "base": base_latency, "candidate": candidate_latency}) + else: + actual = (candidate_latency - base_latency) / base_latency * 100.0 + results.append({"threshold": "max_candidate_mean_latency_regression_pct", "status": "passed" if actual <= latency_limit else "failed", "actual_pct": actual, "limit_pct": latency_limit, "base": base_latency, "candidate": candidate_latency}) + return results + + def require_fresh_work_dirs(targets: list[RunTarget], *, reuse_work_dir: bool, dry_run: bool, fixture_only: bool) -> None: if dry_run or reuse_work_dir or fixture_only: return @@ -1292,6 +1471,68 @@ def run_remote_write_scenario(args: argparse.Namespace, case: dict[str, Any], ca cluster.stop_all() +def run_otlp_trace_load_scenario(args: argparse.Namespace, case: dict[str, Any], targets: list[RunTarget], report: dict[str, Any]) -> None: + load = scenario(case)["load"] + if args.fixture_only: + raise ValueError("--fixture-only is not supported for otlp_trace_load; use --dry-run for planning") + if args.otelgen_bin is not None: + require_binary(args.otelgen_bin, "otelgen", dry_run=args.dry_run) + elif not args.dry_run: + raise ValueError("--otelgen-bin is required for otlp_trace_load") + + clusters: list[DistributedCluster] = [] + metrics_results: list[dict[str, Any]] = [] + try: + for target in targets: + target.work_dir.mkdir(parents=True, exist_ok=True) + cluster = DistributedCluster(target) + clusters.append(cluster) + if not args.dry_run: + cluster.start_all() + create_database: dict[str, Any] = {"status": "dry-run", "database": load["database"]} + if not args.dry_run: + create_database = http_post_sql(target.http_port, f"CREATE DATABASE IF NOT EXISTS {sql_ident(load['database'])}", "public", args.http_timeout) + otelgen = run_otelgen_load(args.otelgen_bin, target, load, args.http_timeout, dry_run=args.dry_run) + metrics: dict[str, Any] = {"status": "dry-run"} + flush: dict[str, Any] = {"status": "dry-run", "table": load["table"]} + visibility: dict[str, Any] = {"status": "dry-run"} + if not args.dry_run: + metrics = summarize_otlp_metrics(otelgen) + flush = http_post_sql(target.http_port, f"ADMIN FLUSH_TABLE({sql_string(load['table'])})", load["database"], args.http_timeout) + visibility = poll_expected_count(target, load["table"], load["database"], int(metrics["accepted_spans"]), float(load["visibility_timeout_seconds"]), args.http_timeout) + tr = { + "name": target.name, + "binary": str(target.binary), + "work_dir": str(target.work_dir), + "components": cluster.component_report(), + "create_database": create_database, + "otelgen": otelgen, + "metrics": metrics, + "flush": flush, + "visibility": visibility, + } + checks_ok = args.dry_run or ( + create_database.get("ok") + and otelgen.get("status") == "ok" + and not metrics.get("missing_metrics") + and metrics.get("accepted_spans", 0) > 0 + and metrics.get("http_requests", 0) > 0 + and flush.get("ok") + and visibility.get("ok") + and visibility.get("row_count_ok") + ) + tr["status"] = "planned" if args.dry_run else ("measured" if checks_ok else "failed") + write_json(target.report_path, tr) + report["targets"].append(tr) + metrics_results.append(metrics) + cluster.stop_all() + report["thresholds"] = planned_otlp_thresholds(load) if args.dry_run else enforce_otlp_thresholds(load, metrics_results[0], metrics_results[1]) + report["status"] = "planned" if args.dry_run else ("failed" if any(t["status"] == "failed" for t in report["thresholds"]) or any(t["status"] == "failed" for t in report["targets"]) else "ok") + finally: + for cluster in reversed(clusters): + cluster.stop_all() + + def main() -> int: args = parse_args() case_path = args.case.resolve() @@ -1314,6 +1555,16 @@ def main() -> int: reuse_fixture = args.reuse_fixture or args.fixture_cache_dir is not None report: dict[str, Any] = {"case_path": str(case_path), "case": case.get("case", {}), "scenario": scenario_config, "queries": planned_queries(case), "dry_run": args.dry_run, "fixture_only": args.fixture_only, "query_mode": "fixture-only" if args.fixture_only else "distributed", "reuse_work_dir": args.reuse_work_dir, "reuse_fixture": reuse_fixture, "fixture_cache_dir": str(args.fixture_cache_dir.resolve()) if args.fixture_cache_dir is not None else None, "fixture_dir": str(fixture_dir), "http_timeout": args.http_timeout, "targets": [], "thresholds": [], "status": "planned" if args.dry_run else "running"} + if scenario_kind == "otlp_trace_load": + try: + run_otlp_trace_load_scenario(args, case, targets, report) + except Exception as e: # noqa: BLE001 - write machine-readable failure report + report["status"] = "failed" + report["error"] = repr(e) + write_json(work_root / "query-regression-report.json", report) + output_report(report, args.output) + return 1 if report["status"] == "failed" else 0 + if scenario_kind == "prom_remote_write_then_query": try: run_remote_write_scenario(args, case, case_path, targets, report) @@ -1321,7 +1572,7 @@ def main() -> int: report["status"] = "failed" report["error"] = repr(e) write_json(work_root / "query-regression-report.json", report) - print(json.dumps(report, indent=2, sort_keys=True)) + output_report(report, args.output) return 1 if report["status"] == "failed" else 0 if args.fixture_only or args.dry_run: @@ -1341,7 +1592,7 @@ def main() -> int: report["targets"].append(tr) report["status"] = "planned" if args.dry_run else "fixture-ready" write_json(work_root / "query-regression-report.json", report) - print(json.dumps(report, indent=2, sort_keys=True)) + output_report(report, args.output) return 0 clusters: list[DistributedCluster] = [] @@ -1404,7 +1655,7 @@ def main() -> int: for cluster in reversed(clusters): cluster.stop_all() write_json(work_root / "query-regression-report.json", report) - print(json.dumps(report, indent=2, sort_keys=True)) + output_report(report, args.output) return 1 if report["status"] == "failed" else 0 diff --git a/tests/perf/test_query_regression_runner_compaction_toctou.py b/tests/perf/test_query_regression_runner_compaction_toctou.py index 675f451a10..0e79cf245d 100644 --- a/tests/perf/test_query_regression_runner_compaction_toctou.py +++ b/tests/perf/test_query_regression_runner_compaction_toctou.py @@ -17,6 +17,7 @@ import argparse import importlib.util +import json import sys import tempfile import unittest @@ -32,6 +33,23 @@ sys.modules[SPEC.name] = runner SPEC.loader.exec_module(runner) +class ReportOutputTest(unittest.TestCase): + def test_final_report_output_defaults_to_stdout_or_writes_to_file(self) -> None: + report = {"status": "ok"} + expected = json.dumps(report, indent=2, sort_keys=True) + + with patch("builtins.print") as print_mock: + runner.output_report(report, None) + print_mock.assert_called_once_with(expected) + + with tempfile.TemporaryDirectory() as tmpdir: + output = Path(tmpdir) / "nested" / "report.json" + with patch("builtins.print") as print_mock: + runner.output_report(report, output) + print_mock.assert_not_called() + self.assertEqual(output.read_text(), expected + "\n") + + class RemoteWriteCompactionToctouTest(unittest.TestCase): def test_quiesces_each_target_before_storage_inspection_and_read_bench(self) -> None: lifecycle: dict[str, list[str]] = {"base": [], "candidate": []} diff --git a/tests/perf/test_query_regression_runner_otlp_trace_load.py b/tests/perf/test_query_regression_runner_otlp_trace_load.py new file mode 100644 index 0000000000..ab14ceef24 --- /dev/null +++ b/tests/perf/test_query_regression_runner_otlp_trace_load.py @@ -0,0 +1,228 @@ +#!/usr/bin/env python3 +# Copyright 2023 Greptime Team +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Regression coverage for the local OTLP trace load lifecycle and metrics.""" + +import importlib.util +import json +import os +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + + +RUNNER_PATH = Path(__file__).with_name("query_regression_runner.py") +PLOTTER_PATH = Path(__file__).with_name("plot_otlp_trace_report.sh") +SPEC = importlib.util.spec_from_file_location("query_regression_runner_otlp_under_test", RUNNER_PATH) +assert SPEC is not None and SPEC.loader is not None +runner = importlib.util.module_from_spec(SPEC) +sys.modules[SPEC.name] = runner +SPEC.loader.exec_module(runner) + + +class OtlpTraceReportPlotTest(unittest.TestCase): + def test_plots_all_comparison_metrics_and_thresholds(self) -> None: + report = { + "status": "ok", + "targets": [ + { + "name": name, + "metrics": { + "accepted_spans": spans, + "accepted_spans_per_second": rate, + "mean_http_latency_ms": latency, + "failure_count": 0, + }, + "visibility": {"observed_rows": spans}, + } + for name, spans, rate, latency in (("base", 100, 10.0, 3.0), ("candidate", 120, 12.0, 2.5)) + ], + "thresholds": [ + {"target": "base", "threshold": "max_failure_count", "status": "passed", "actual": 0, "limit": 0}, + { + "threshold": "max_candidate_throughput_regression_pct", + "status": "passed", + "actual_pct": -20.0, + "limit_pct": 20.0, + "base": 10.0, + "candidate": 12.0, + }, + ], + } + + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + report_path = root / "report.json" + report_path.write_text(json.dumps(report)) + fake_bin = root / "bin" + fake_bin.mkdir() + uplot = fake_bin / "uplot" + uplot.write_text('#!/usr/bin/env bash\nprintf "CALL %s\\n" "$*" >>"$UPLOT_LOG"\ncat >>"$UPLOT_LOG"\n') + uplot.chmod(0o755) + log_path = root / "uplot.log" + env = os.environ.copy() + env["PATH"] = f"{fake_bin}{os.pathsep}{env['PATH']}" + env["UPLOT_LOG"] = str(log_path) + + result = subprocess.run([str(PLOTTER_PATH), str(report_path)], env=env, text=True, capture_output=True, check=False) + log = log_path.read_text() + report["targets"][1]["visibility"]["observed_rows"] = 119 + report_path.write_text(json.dumps(report)) + mismatch = subprocess.run([str(PLOTTER_PATH), str(report_path)], env=env, text=True, capture_output=True, check=False) + + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual(log.count("CALL "), 5) + for metric in ("Accepted spans", "Visible table rows", "Throughput", "Mean HTTP latency", "Failures"): + self.assertIn(metric, log) + self.assertIn("base\t100", log) + self.assertIn("candidate\t12", log) + self.assertIn("Status: ok", result.stdout) + self.assertIn("[passed] candidate: accepted=120, visible=120", result.stdout) + self.assertIn("max_failure_count", result.stdout) + self.assertIn("max_candidate_throughput_regression_pct", result.stdout) + self.assertEqual(mismatch.returncode, 1, mismatch.stderr) + self.assertIn("[failed] candidate: accepted=120, visible=119", mismatch.stdout) + + +class OtlpTraceLoadTest(unittest.TestCase): + def test_stops_base_cluster_before_creating_candidate_cluster(self) -> None: + events = [] + + class FakeCluster: + def __init__(self, target): + self.target = target + self.stopped = False + events.append(f"create:{target.name}") + + def component_report(self): + return {} + + def stop_all(self): + if not self.stopped: + self.stopped = True + events.append(f"stop:{self.target.name}") + + load = { + "database": "public", + "table": "opentelemetry_traces", + "pipeline": "greptime_trace_v1", + "duration_seconds": 120, + "warmup_seconds": 60, + "rate": 50_000, + "workers": 4, + "workload": "microservices", + "exporter_shards": 4, + "visibility_timeout_seconds": 1, + "thresholds": { + "max_candidate_throughput_regression_pct": 20, + "max_candidate_mean_latency_regression_pct": 20, + "max_failure_count": 0, + }, + } + args = runner.argparse.Namespace(fixture_only=False, otelgen_bin=None, dry_run=True, http_timeout=1.0) + + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + targets = [ + runner.make_target("base", Path("/bin/true"), root, list(range(10_000, 10_008))), + runner.make_target("candidate", Path("/bin/true"), root, list(range(10_008, 10_016))), + ] + with patch.object(runner, "DistributedCluster", FakeCluster): + runner.run_otlp_trace_load_scenario( + args, + {"scenario": {"kind": "otlp_trace_load", "load": load}}, + targets, + {"targets": []}, + ) + + self.assertEqual(events, ["create:base", "stop:base", "create:candidate", "stop:candidate"]) + + def test_warmup_lifecycle_and_labeled_metric_deltas(self) -> None: + class FakeProcess: + def __init__(self, *_args, **_kwargs): + self.returncode = None + self.wait_timeouts = [] + + def wait(self, timeout): + self.wait_timeouts.append(timeout) + if len(self.wait_timeouts) == 1: + raise subprocess.TimeoutExpired("otelgen", timeout) + self.returncode = 0 + return 0 + + def poll(self): + return self.returncode + + def kill(self): + self.returncode = -9 + + def snapshot(text: str, captured: float): + return {"captured_monotonic_seconds": captured, "values": runner.parse_prometheus_metrics(text)} + + snapshots = [ + snapshot("greptime_frontend_otlp_traces_rows 10\n", 0.0), + snapshot( + 'greptime_frontend_otlp_traces_rows 110\n' + 'greptime_servers_http_otlp_traces_elapsed_sum{db="public"} 1\n' + 'greptime_servers_http_otlp_traces_elapsed_count{db="public"} 10\n', + 5.0, + ), + snapshot( + 'greptime_frontend_otlp_traces_rows 310\n' + 'greptime_frontend_otlp_traces_failure_count{label="decode"} 1\n' + 'greptime_frontend_otlp_traces_failure_count{label="write"} 2\n' + 'greptime_servers_http_otlp_traces_elapsed_sum{db="public"} 3\n' + 'greptime_servers_http_otlp_traces_elapsed_count{db="public"} 30\n', + 15.0, + ), + ] + load = { + "database": "public", + "table": "opentelemetry_traces", + "pipeline": "greptime_trace_v1", + "duration_seconds": 120, + "warmup_seconds": 60, + "rate": 50_000, + "workers": 4, + "workload": "microservices", + "exporter_shards": 4, + } + + with tempfile.TemporaryDirectory() as tmpdir: + target = runner.make_target("base", Path("/bin/true"), Path(tmpdir), list(range(10_000, 10_008))) + process = FakeProcess() + with ( + patch.object(runner, "fetch_otlp_metrics", side_effect=snapshots), + patch.object(runner.subprocess, "Popen", return_value=process), + patch.object(runner.time, "monotonic", side_effect=[0.0, 120.0]), + ): + result = runner.run_otelgen_load(Path("/bin/otelgen"), target, load, 1.0, dry_run=False) + + self.assertEqual(result["status"], "ok") + self.assertEqual(process.wait_timeouts, [60, 120]) + metrics = runner.summarize_otlp_metrics(result) + self.assertEqual(metrics["accepted_spans"], 300) + self.assertEqual(metrics["measurement_accepted_spans"], 200) + self.assertEqual(metrics["accepted_spans_per_second"], 20.0) + self.assertEqual(metrics["http_requests"], 20) + self.assertEqual(metrics["mean_http_latency_ms"], 100.0) + self.assertEqual(metrics["failure_count"], 3) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/perf/test_query_regression_summary_otlp.py b/tests/perf/test_query_regression_summary_otlp.py new file mode 100644 index 0000000000..e0c91088d8 --- /dev/null +++ b/tests/perf/test_query_regression_summary_otlp.py @@ -0,0 +1,88 @@ +#!/usr/bin/env python3 +# Copyright 2023 Greptime Team +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Regression coverage for the OTLP trace GitHub summary.""" + +import json +import subprocess +import sys +import tempfile +import unittest +from pathlib import Path + + +SUMMARY = Path(__file__).parents[2] / ".github/scripts/query-regression-summary.py" + + +class OtlpTraceSummaryTest(unittest.TestCase): + def test_renders_ingestion_metrics_and_thresholds(self) -> None: + report = { + "status": "ok", + "case": {"name": "otlp_trace_load"}, + "scenario": {"kind": "otlp_trace_load"}, + "targets": [ + { + "name": "base", + "status": "measured", + "metrics": { + "accepted_spans": 100, + "accepted_spans_per_second": 200.0, + "mean_http_latency_ms": 10.0, + "failure_count": 0, + }, + "visibility": {"observed_rows": 100}, + }, + { + "name": "candidate", + "status": "measured", + "metrics": { + "accepted_spans": 90, + "accepted_spans_per_second": 180.0, + "mean_http_latency_ms": 11.0, + "failure_count": 0, + }, + "visibility": {"observed_rows": 90}, + }, + ], + "thresholds": [ + { + "threshold": "max_candidate_throughput_regression_pct", + "actual_pct": 10.0, + "limit_pct": 20.0, + "status": "passed", + } + ], + } + with tempfile.TemporaryDirectory() as tmpdir: + report_path = Path(tmpdir) / "query-regression-report.json" + report_path.write_text(json.dumps(report)) + result = subprocess.run( + [sys.executable, str(SUMMARY), "--report", str(report_path)], + check=True, + capture_output=True, + text=True, + ) + + self.assertIn("### OTLP trace comparison", result.stdout) + self.assertIn("| base | 100 | 100 | 200.00 | 10.00 | 0 |", result.stdout) + self.assertIn( + "| max_candidate_throughput_regression_pct | base vs candidate | 10.00% | 20.00% | โœ… passed | N/A |", + result.stdout, + ) + self.assertNotIn("### Query comparison", result.stdout) + + +if __name__ == "__main__": + unittest.main()