mirror of
https://github.com/GreptimeTeam/greptimedb.git
synced 2026-08-18 12:08:22 +00:00
ci: add OTLP trace ingestion regression testing (#8631)
* chore: update CI config Signed-off-by: shuiyisong <xixing.sys@gmail.com> * chore: add CI Signed-off-by: shuiyisong <xixing.sys@gmail.com> * chore: update CI config Signed-off-by: shuiyisong <xixing.sys@gmail.com> * ci: report otelgen runner diagnostics Signed-off-by: shuiyisong <xixing.sys@gmail.com> * chore: add script to draw result diagram Signed-off-by: shuiyisong <xixing.sys@gmail.com> --------- Signed-off-by: shuiyisong <xixing.sys@gmail.com>
This commit is contained in:
shuiyisong
GitHub
@@ -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 \
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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"))
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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="<unavailable>"
|
||||
if [[ -n "${otelgen_path}" ]]; then
|
||||
otelgen_version="$("${otelgen_path}" --version 2>&1 || true)"
|
||||
fi
|
||||
printf 'otelgen path: %s\n' "${otelgen_path:-<missing>}"
|
||||
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
|
||||
|
||||
@@ -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",
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -125,6 +125,28 @@ fn run_plan(args: PlanArgs) -> Result<(), Box<dyn std::error::Error>> {
|
||||
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}))?
|
||||
|
||||
@@ -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 <workflow-branch> \
|
||||
-f case=tests/perf/query_cases/otlp_trace_load/case.toml \
|
||||
-f base_ref=<full-base-sha> \
|
||||
-f candidate_ref=<full-candidate-sha> \
|
||||
-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:
|
||||
|
||||
Executable
+80
@@ -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}"
|
||||
@@ -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
|
||||
@@ -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
|
||||
|
||||
|
||||
|
||||
@@ -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": []}
|
||||
|
||||
@@ -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()
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user