mirror of
https://github.com/GreptimeTeam/greptimedb.git
synced 2026-10-12 15:02:32 +00:00
ci: enhance benchmark workflows with version comparisons and diagnostics (#9383)
* ci: update Agent and Traces benchmark runtime Signed-off-by: WenyXu <wenymedia@gmail.com> * ci: use unrestricted benchmark runtimes by default Signed-off-by: WenyXu <wenymedia@gmail.com> * ci: retry benchmark image pulls before generating datasets Signed-off-by: WenyXu <wenymedia@gmail.com> * ci: increase tracesbench defaults to 8 CPU and 32 GiB Signed-off-by: WenyXu <wenymedia@gmail.com> * ci: configure tracesbench query timeout and update runtime Signed-off-by: WenyXu <wenymedia@gmail.com> * ci: update tracesbench runtime and configure import scan workers Signed-off-by: WenyXu <wenymedia@gmail.com> * ci: use shared parallel tracesbench import preparation Signed-off-by: WenyXu <wenymedia@gmail.com> * ci: update tracesbench runtime to preserve Tempo trace requests Signed-off-by: WenyXu <wenymedia@gmail.com> * ci: summarize tracesbench query results and lifecycle Signed-off-by: WenyXu <wenymedia@gmail.com> * ci: show benchmark resource usage and update parallel loader runtime Signed-off-by: WenyXu <wenymedia@gmail.com> * fix(ci): resolve shared summary module in existing test harness Signed-off-by: WenyXu <wenymedia@gmail.com> * ci: select tracesbench engines and GreptimeDB image version Signed-off-by: WenyXu <wenymedia@gmail.com> * ci: update benchmark runtime with parallel Agent preparation Signed-off-by: WenyXu <wenymedia@gmail.com> * ci: select benchmark runtime registry by cloud provider Signed-off-by: WenyXu <wenymedia@gmail.com> * ci: retry long-range benchmark image pulls Signed-off-by: WenyXu <wenymedia@gmail.com> * ci: compare baseline and candidate GreptimeDB benchmarks Signed-off-by: WenyXu <wenymedia@gmail.com> * ci: pin published dual-version benchmark runtime Signed-off-by: WenyXu <wenymedia@gmail.com> * ci: prepare agent benchmark indexes with updated runtime Signed-off-by: WenyXu <wenymedia@gmail.com> * ci: validate agent admission and render traces version comparison Signed-off-by: WenyXu <wenymedia@gmail.com> * ci: use runtime with traces index preparation Signed-off-by: WenyXu <wenymedia@gmail.com> * ci: use consolidated runtime for benchmark version reports Signed-off-by: WenyXu <wenymedia@gmail.com> * test: add license header to benchmark comparison tests Signed-off-by: WenyXu <wenymedia@gmail.com> * fix(ci): read legacy Tempo ingestion metrics in traces summaries Signed-off-by: WenyXu <wenymedia@gmail.com> * ci: require tracesbench query plan artifacts Signed-off-by: WenyXu <wenymedia@gmail.com> * ci: use optimized tracesbench report runtime Signed-off-by: WenyXu <wenymedia@gmail.com> * ci: update tracesbench runtime to merged report validation Signed-off-by: WenyXu <wenymedia@gmail.com> --------- Signed-off-by: WenyXu <wenymedia@gmail.com>
This commit is contained in:
@@ -18,6 +18,8 @@ import json
|
||||
import math
|
||||
from pathlib import Path
|
||||
|
||||
from benchmark_resource_summary import render_resources
|
||||
|
||||
TARGETS = ('greptimedb', 'clickhouse', 'victorialogs')
|
||||
|
||||
|
||||
@@ -201,6 +203,7 @@ def lifecycle(root, targets):
|
||||
rows.append([context, entry.get('phase'), duration(entry.get('elapsed_ms')) if valid else
|
||||
'invalid timestamp units (legacy artifact)', entry.get('exit_code')])
|
||||
lines += ['## Phase timings', '', 'Phase duration includes orchestration overhead. Container cleanup does not prove ECS teardown.', '', table(['DB / scope', 'Phase', 'Duration', 'Exit code'], rows), '']
|
||||
lines.append(render_resources([('dataset', root)] + [(t, root / f'{t}-1') for t in targets], table))
|
||||
return '\n'.join(lines)
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,191 @@
|
||||
# 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.
|
||||
|
||||
"""Compare validated GreptimeDB query results; never gate CI on noisy latency deltas."""
|
||||
import argparse
|
||||
import html
|
||||
import json
|
||||
import math
|
||||
import os
|
||||
from pathlib import Path
|
||||
import re
|
||||
import statistics
|
||||
|
||||
|
||||
def read(path):
|
||||
return json.loads(path.read_text()) if path.is_file() else {}
|
||||
|
||||
|
||||
def validate(env):
|
||||
enabled = env.get('COMPARE_GREPTIMEDB', 'false')
|
||||
if enabled not in ('true', 'false'):
|
||||
raise ValueError('COMPARE_GREPTIMEDB must be true or false')
|
||||
if enabled == 'false':
|
||||
return
|
||||
if env.get('RUN_GREPTIMEDB') != 'true':
|
||||
raise ValueError('Version comparison requires Run GreptimeDB')
|
||||
for key in ('BASELINE_GREPTIMEDB_TAG', 'GREPTIMEDB_TAG'):
|
||||
if not re.fullmatch(r'[a-zA-Z0-9_][a-zA-Z0-9_.-]{0,127}(@sha256:[0-9a-f]{64})?', env.get(key, '')):
|
||||
raise ValueError(f'Invalid {key}')
|
||||
if env['BASELINE_GREPTIMEDB_TAG'] == env['GREPTIMEDB_TAG']:
|
||||
raise ValueError('Baseline and candidate must have different image tags')
|
||||
|
||||
|
||||
def same(a, b, fields):
|
||||
return all(a.get(k) is not None and a.get(k) == b.get(k) for k in fields)
|
||||
|
||||
|
||||
def finite(value):
|
||||
return isinstance(value, (int, float)) and not isinstance(value, bool) and math.isfinite(value) and value >= 0
|
||||
|
||||
|
||||
def improvement(old, new, throughput=False):
|
||||
if not finite(old) or not finite(new) or old <= 0:
|
||||
return None
|
||||
return (new / old - 1 if throughput else 1 - new / old) * 100
|
||||
|
||||
|
||||
def row(query, metric, old, new, compatible, reason):
|
||||
delta = improvement(old, new, metric == 'QPS') if compatible else None
|
||||
return dict(query=query, metric=metric, baseline=old, candidate=new,
|
||||
improvement_percent=delta, status='comparable' if delta is not None else reason)
|
||||
|
||||
|
||||
def agent(root):
|
||||
def load(r):
|
||||
return read(r / 'greptimedb-1/greptimedb-1/measured-summary.json')
|
||||
a, b = load(root / 'baseline'), load(root)
|
||||
finished = [r / 'greptimedb-1/exit-code.txt' for r in (root / 'baseline', root)]
|
||||
compatible = (all(p.is_file() and p.read_text().strip() == '0' for p in finished)
|
||||
and a.get('passed') is True and b.get('passed') is True
|
||||
and same(a, b, ('contract_sha256', 'data_profile', 'query_window', 'warmup_runs', 'measured_runs')))
|
||||
qa, qb = ({q['query_id']: q for q in x.get('per_query', [])} for x in (a, b))
|
||||
rows = []
|
||||
for q in sorted(qa.keys() | qb.keys()):
|
||||
old, new = qa.get(q, {}), qb.get(q, {})
|
||||
ok = compatible and same(old, new, ('result_fingerprint', 'success')) and old.get('success', 0) > 0
|
||||
ok = ok and all(v.get('errors') == 0 and v.get('timeouts') == 0 for v in (old, new))
|
||||
for key in ('p50_ms', 'p95_ms', 'p99_ms'):
|
||||
rows.append(row(q, key, old.get(key), new.get(key), ok, 'failed, missing or incompatible result'))
|
||||
return rows
|
||||
|
||||
|
||||
def traces(root):
|
||||
a, b = (read(r / 'runs/greptimedb/run.json') for r in (root / 'baseline', root))
|
||||
compatible = (a.get('success') is True and b.get('success') is True
|
||||
and same(a, b, ('expected', 'protocol', 'audit_expected', 'corpus'))
|
||||
and same(a.get('execution', {}), b.get('execution', {}),
|
||||
('db_cpus', 'db_memory', 'load_workers_requested', 'max_in_flight')))
|
||||
rows = []
|
||||
for q in (f'T{i:02}' for i in range(1, 9)):
|
||||
values, identities, counts = [], [], []
|
||||
ok = compatible
|
||||
for run in (a, b):
|
||||
correctness = [s for s in run.get('correctness', []) if s.get('query_id') == q]
|
||||
all_samples = [s for s in run.get('samples', []) if s.get('query_id') == q]
|
||||
measured = [s for s in all_samples if s.get('phase') == 'measured']
|
||||
passed = bool(correctness and measured) and all(s.get('outcome') == 'passed' for s in correctness + all_samples)
|
||||
expected = run.get('expected', {}).get('contracts', {}).get(q, {})
|
||||
passed = passed and all(s.get('fingerprint') == expected.get('fingerprint') and s.get('fingerprint') is not None
|
||||
and s.get('row_count') == expected.get('row_count') for s in correctness + all_samples)
|
||||
elapsed = [s.get('elapsed_ms') for s in measured]
|
||||
passed = passed and all(finite(v) for v in elapsed)
|
||||
ok = ok and passed
|
||||
values.append(statistics.median(elapsed) if passed else None)
|
||||
identities.append(expected)
|
||||
counts.append((len(correctness), len(all_samples), len(measured)))
|
||||
ok = ok and identities[0] == identities[1] and counts[0] == counts[1]
|
||||
rows.append(row(q, 'median_ms', *values, ok, 'failed, skipped, missing or incompatible result'))
|
||||
return rows
|
||||
|
||||
|
||||
def long_range(root):
|
||||
rows = []
|
||||
manifest = read(root / 'run-manifest.json')
|
||||
healthy = all(
|
||||
all(field in (manifest.get(phase, {}).get(target, {}).get('docker', {}) if phase == 'post_load_results'
|
||||
else manifest.get(phase, {}).get(target, {})).get('state', '').split()
|
||||
for field in ('Running=true', 'ExitCode=0', 'OOMKilled=false'))
|
||||
for phase in ('post_load_results', 'post_query_results')
|
||||
for target in ('greptimedb-baseline', 'greptimedb'))
|
||||
for mode in ('serial', 'concurrent-duration-c1', 'concurrent-duration-c4'):
|
||||
a, b = (read(root / name / 'results-vmbench-long-range' / (mode + '-summary.json'))
|
||||
for name in ('greptimedb-baseline', 'greptimedb'))
|
||||
fields = ('workload', 'query_file_sha256', 'start', 'end', 'step', 'profile', 'data_duration',
|
||||
'scrape_interval', 'cache_policy_effective', 'mode', 'connection_mode', 'request_timeout_seconds')
|
||||
fields += ('tries', 'warmup_runs') if mode == 'serial' else ('concurrency', 'duration_seconds', 'warmup_seconds')
|
||||
compatible = healthy and same(a, b, fields) and all(x.get('errors') == 0 and x.get('timeouts') == 0 for x in (a, b))
|
||||
qa, qb = ({q['query_index']: q for q in x.get('per_query', [])} for x in (a, b))
|
||||
for q in sorted(qa.keys() | qb.keys()):
|
||||
old, new = qa.get(q, {}), qb.get(q, {})
|
||||
ok = compatible and same(old, new, ('query', 'series', 'points'))
|
||||
ok = ok and all(v.get('success', 0) > 0 and v.get('series', 0) > 0 and v.get('points', 0) > 0
|
||||
and v.get('errors') == 0 and v.get('timeouts') == 0 for v in (old, new))
|
||||
for key in ('p50_ms', 'qps') if mode != 'serial' else ('p50_ms',):
|
||||
rows.append(row(f'{mode}/Q{q}', 'QPS' if key == 'qps' else key,
|
||||
old.get(key), new.get(key), ok, 'failed, missing or incompatible result'))
|
||||
return rows
|
||||
|
||||
|
||||
def render(document):
|
||||
def cell(value):
|
||||
return html.escape(str(value)).replace('|', '|').replace('\n', ' ')
|
||||
def value(v, metric):
|
||||
if not finite(v):
|
||||
return '—'
|
||||
if metric == 'QPS':
|
||||
return f'{v:.3f}'
|
||||
return f'{v / 1000:.2f} s' if v >= 1000 else f'{v:.2f} ms'
|
||||
lines = ['# GreptimeDB version comparison', '',
|
||||
f'Baseline: `{cell(document["baseline_tag"])}` → candidate: `{cell(document["candidate_tag"])}`.', '',
|
||||
'Same-host sequential runs with fresh databases and shared input data. Positive = improvement; '
|
||||
'latency: (baseline − candidate) / baseline; QPS: (candidate − baseline) / baseline. '
|
||||
'Single runs are diagnostic, not a statistically established regression. No performance threshold gates CI.', '',
|
||||
'| Query | Metric | Baseline | Candidate | Improvement | Status |',
|
||||
'| --- | --- | --- | --- | --- | --- |']
|
||||
for r in document['rows']:
|
||||
delta = r['improvement_percent']
|
||||
lines.append('| ' + ' | '.join(map(cell, [r['query'], r['metric'], value(r['baseline'], r['metric']),
|
||||
value(r['candidate'], r['metric']), f'{delta:+.1f}%' if delta is not None else '—', r['status']])) + ' |')
|
||||
if document.get('kind') == 'long-range':
|
||||
lines += ['', 'Long-range retains the existing non-empty series/point-count checks; these do not establish equality of every returned value.']
|
||||
if not document['rows']:
|
||||
lines += ['', 'Results unavailable; comparison not performed.']
|
||||
return '\n'.join(lines) + '\n'
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument('command', choices=('validate', 'report'))
|
||||
parser.add_argument('--kind', choices=('agent', 'traces', 'long-range'))
|
||||
parser.add_argument('--root', type=Path)
|
||||
args = parser.parse_args()
|
||||
if args.command == 'validate':
|
||||
validate(os.environ)
|
||||
return
|
||||
if args.root is None or args.kind is None:
|
||||
parser.error('report requires --root and --kind')
|
||||
document = {'kind': args.kind, 'baseline_tag': os.environ.get('BASELINE_GREPTIMEDB_TAG', ''),
|
||||
'candidate_tag': os.environ.get('GREPTIMEDB_TAG', ''),
|
||||
'rows': {'agent': agent, 'traces': traces, 'long-range': long_range}[args.kind](args.root)}
|
||||
text = render(document)
|
||||
# Preserve partial comparisons even if a target failed before creating output.
|
||||
args.root.mkdir(parents=True, exist_ok=True)
|
||||
(args.root / 'version-comparison.json').write_text(json.dumps(document, indent=2, allow_nan=False) + '\n')
|
||||
(args.root / 'version-comparison.md').write_text(text)
|
||||
print(text, end='')
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
@@ -0,0 +1,99 @@
|
||||
# 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.
|
||||
|
||||
"""Render shared O11yBench sampled resource artifacts for lifecycle summaries."""
|
||||
import json
|
||||
import math
|
||||
|
||||
|
||||
def number(value):
|
||||
return isinstance(value, (int, float)) and not isinstance(value, bool) and math.isfinite(value) and value >= 0
|
||||
|
||||
|
||||
def metric(values, key, statistic):
|
||||
entry = values.get(key, {})
|
||||
value = entry.get(statistic)
|
||||
return value if number(entry.get('samples')) and entry['samples'] > 0 and number(value) else None
|
||||
|
||||
|
||||
def decimal(value, suffix=''):
|
||||
return f'{value:.2f}{suffix}' if number(value) else '—'
|
||||
|
||||
|
||||
def memory(value):
|
||||
if not number(value):
|
||||
return '—'
|
||||
for unit in ('B', 'KiB', 'MiB', 'GiB', 'TiB'):
|
||||
if value < 1024 or unit == 'TiB':
|
||||
return decimal(value, f' {unit}')
|
||||
value /= 1024
|
||||
|
||||
|
||||
def capacities(path):
|
||||
# Quotas are metadata, not recomputed CPU statistics. Bound reads on long runs.
|
||||
result = {}
|
||||
try:
|
||||
with path.open() as source:
|
||||
for index, line in enumerate(source):
|
||||
if index == 256:
|
||||
break
|
||||
record = json.loads(line)
|
||||
cores = record.get('host', {}).get('cpu_count')
|
||||
if number(cores) and cores > 0:
|
||||
result.setdefault('host', cores)
|
||||
for role, values in record.get('containers', {}).items():
|
||||
quota = values.get('configured_cpu_limit_cores')
|
||||
if number(quota):
|
||||
result.setdefault(role, quota)
|
||||
if all(role in result for role in ('host', 'runtime', 'db')):
|
||||
break
|
||||
except (OSError, ValueError, TypeError, AttributeError):
|
||||
pass
|
||||
return result
|
||||
|
||||
|
||||
def render_resources(scopes, table):
|
||||
rows = []
|
||||
for scope, directory in scopes:
|
||||
try:
|
||||
summary = json.loads((directory / 'resources.summary.json').read_text())
|
||||
phases = summary.get('phases', {}) if summary.get('schema') == 'o11ybench-resources-v1' else {}
|
||||
except (OSError, ValueError, TypeError, AttributeError):
|
||||
phases = {}
|
||||
if not isinstance(phases, dict) or not phases:
|
||||
rows.append([scope, 'unavailable', *['—'] * 9])
|
||||
continue
|
||||
limits = capacities(directory / 'resources.jsonl')
|
||||
for phase, data in phases.items():
|
||||
for role in ('host', 'runtime', 'db'):
|
||||
values = data.get('host', {}) if role == 'host' else data.get('containers', {}).get(role, {})
|
||||
if role == 'db' and scope == 'dataset':
|
||||
continue
|
||||
quota = limits.get(role)
|
||||
cpu_samples = values.get('cpu_used_cores', {}).get('samples')
|
||||
rows.append([scope, phase, role, cpu_samples,
|
||||
decimal(metric(values, 'cpu_used_cores', 'mean')),
|
||||
decimal(metric(values, 'cpu_used_cores', 'max')),
|
||||
'unlimited' if quota == 0 else decimal(quota),
|
||||
memory(metric(values, 'memory_used_bytes' if role == 'host' else 'memory_usage_bytes', 'max')),
|
||||
decimal(metric(values, 'io_wait_percent', 'mean'), '%'),
|
||||
decimal(metric(values, 'io_wait_percent', 'max'), '%'),
|
||||
memory(metric(values, 'swap_used_bytes', 'max'))])
|
||||
return '\n'.join(['## Sampled resources', '',
|
||||
'CPU is measured in cores. Mean/max summarize available samples, not continuous peaks; '
|
||||
'short phases may have no CPU sample. CPU capacity is host cores or the configured container quota. '
|
||||
'Host memory is total minus available; Docker memory excludes cache where supported. '
|
||||
'Swap is host-wide usage, not swap activity. Missing evidence stays —.', '',
|
||||
table(['Scope', 'Phase', 'Component', 'CPU samples', 'Mean CPU cores', 'Max CPU cores',
|
||||
'CPU capacity', 'Max memory', 'Mean I/O wait', 'Max I/O wait', 'Max swap'], rows), ''])
|
||||
@@ -0,0 +1,25 @@
|
||||
#!/usr/bin/env bash
|
||||
# Pull only: never retry database writes, benchmarks, or cloud provisioning.
|
||||
set -euo pipefail
|
||||
|
||||
if [[ $# != 1 || -z "$1" || "$1" == -* ]]; then
|
||||
echo "Usage: $0 IMAGE" >&2
|
||||
exit 2
|
||||
fi
|
||||
image=$1
|
||||
attempts=3
|
||||
for ((attempt=1; attempt<=attempts; attempt++)); do
|
||||
echo "Pulling $image (attempt $attempt/$attempts)" >&2
|
||||
if timeout --kill-after=10s 180s docker pull "$image"; then
|
||||
exit 0
|
||||
else
|
||||
status=$?
|
||||
fi
|
||||
if ((attempt == attempts)); then
|
||||
echo "Failed to pull $image after $attempts attempts (exit $status)" >&2
|
||||
exit "$status"
|
||||
fi
|
||||
delay=$((attempt * 10))
|
||||
echo "Pull failed (exit $status); retrying in ${delay}s" >&2
|
||||
sleep "$delay"
|
||||
done
|
||||
@@ -0,0 +1,169 @@
|
||||
# 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.
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import html
|
||||
import math
|
||||
from pathlib import Path
|
||||
|
||||
from benchmark_resource_summary import render_resources
|
||||
|
||||
TARGETS = ('greptimedb', 'victoriatraces', 'tempo')
|
||||
|
||||
|
||||
def duration(ms):
|
||||
if not isinstance(ms, (int, float)) or not math.isfinite(ms) or ms < 0:
|
||||
return '—'
|
||||
if ms < 1:
|
||||
return f'{ms * 1000:.0f} µs'
|
||||
if ms < 1000:
|
||||
return f'{ms:.1f} ms'
|
||||
if ms < 60_000:
|
||||
return f'{ms / 1000:.2f} s'
|
||||
minutes, seconds = divmod(round(ms / 1000), 60)
|
||||
hours, minutes = divmod(minutes, 60)
|
||||
return f'{hours}h {minutes}m {seconds}s' if hours else f'{minutes}m {seconds}s'
|
||||
|
||||
|
||||
def cell(value):
|
||||
return html.escape(str(value if value is not None else '—')).replace('|', '|').replace('\n', ' ')
|
||||
|
||||
|
||||
def read(path):
|
||||
if not path.exists():
|
||||
return {}
|
||||
return json.loads(path.read_text())
|
||||
|
||||
|
||||
def table(headers, rows):
|
||||
return '\n'.join(['| ' + ' | '.join(map(cell, headers)) + ' |',
|
||||
'| ' + ' | '.join('---' for _ in headers) + ' |'] +
|
||||
['| ' + ' | '.join(map(cell, row)) + ' |' for row in rows])
|
||||
|
||||
|
||||
def delta(value, baseline):
|
||||
if not all(isinstance(v, (int, float)) and math.isfinite(v) for v in (value, baseline)):
|
||||
return '—'
|
||||
return f'{(value / baseline - 1) * 100:+.1f}%' if baseline > 0 and value >= 0 else '—'
|
||||
|
||||
|
||||
|
||||
def load(root):
|
||||
report = read(root / 'report.json')
|
||||
runs = [read(p) for p in sorted((root / 'runs').glob('*/run.json'))]
|
||||
return report, runs
|
||||
|
||||
|
||||
def benchmark(report, runs, targets=TARGETS):
|
||||
lines = ['# Tracesbench benchmark summary', '',
|
||||
'Median Δ = (target / GreptimeDB − 1) × 100%; negative is faster. '
|
||||
'Min/median/max come from the validated report, not recalculated samples. '
|
||||
'Counts exclude warmup. Missing or excluded results have no latency comparison.',
|
||||
'', f'Classification: {cell(report.get("classification"))}. '
|
||||
'See the uploaded HTML for query semantics and client-side compensation.', '']
|
||||
statistics = {}
|
||||
for corpus in report.get('per_corpus', []):
|
||||
for stat in corpus.get('statistics', []):
|
||||
if stat['target'] not in targets:
|
||||
continue
|
||||
statistics[(corpus['corpus'], stat['round'], stat['target'], stat['query_id'])] = stat
|
||||
rows, details = [], []
|
||||
for target in targets:
|
||||
selected = [run for run in runs if run.get('target') == target]
|
||||
if not selected:
|
||||
lines.append(f'- **{target}**: unavailable / not run.')
|
||||
for run in selected:
|
||||
lines.append(f'- **{target}**: lifecycle '
|
||||
f'{"completed" if run.get("success") is True else "failed / incomplete"}; '
|
||||
f'query failures: {len(run.get("query_failures", []))}.')
|
||||
for qid in (f'T{i:02}' for i in range(1, 9)):
|
||||
correct = [q for q in run.get('correctness', []) if q.get('query_id') == qid]
|
||||
measured = [q for q in run.get('samples', [])
|
||||
if q.get('query_id') == qid and q.get('phase') == 'measured']
|
||||
errors = [q for q in correct + run.get('samples', [])
|
||||
if q.get('query_id') == qid and q.get('outcome') not in ('passed', 'not_executed')]
|
||||
reason = (run.get('exclusions', {}).get(qid) or
|
||||
run.get('execution', {}).get('query_skips', {}).get(qid))
|
||||
stat = statistics.get((run.get('corpus'), run.get('round'), target, qid), {})
|
||||
base = statistics.get((run.get('corpus'), run.get('round'), 'greptimedb', qid), {})
|
||||
status = ('failed' if errors else 'excluded' if reason else
|
||||
'passed' if stat else
|
||||
'correctness passed; report unavailable' if correct and
|
||||
all(q.get('outcome') == 'passed' for q in correct) else 'unavailable / incomplete')
|
||||
rows.append([run.get('corpus'), run.get('round'), qid, target, status,
|
||||
duration(stat.get('min_ms')), duration(stat.get('median_ms')),
|
||||
duration(stat.get('max_ms')),
|
||||
delta(stat.get('median_ms'), base.get('median_ms')) if target != 'greptimedb' else '—',
|
||||
sum(q.get('outcome') == 'passed' for q in measured),
|
||||
sum(q.get('outcome') != 'passed' for q in measured)])
|
||||
if reason or errors:
|
||||
messages = [str(q.get('error') or q.get('outcome')) for q in errors]
|
||||
details.append(f'- **{target} / {qid}**: {cell(str(reason or "") + " " + "; ".join(messages))[:4000]}')
|
||||
lines += ['', table(['Dataset', 'Round', 'Query', 'DB', 'Status', 'Min', 'Median', 'Max',
|
||||
'Median Δ', 'Measured success', 'Measured errors'], rows), '',
|
||||
'<details>', '<summary>Query failures and exclusions</summary>', '', *details, '', '</details>']
|
||||
if not report:
|
||||
lines += ['', 'Validated report unavailable; raw outcomes are shown without latency statistics.']
|
||||
return '\n'.join(lines)
|
||||
|
||||
|
||||
def lifecycle(report, runs, root, targets=TARGETS):
|
||||
rows = []
|
||||
for target in targets:
|
||||
selected = [r for r in runs if r.get('target') == target] or [{}]
|
||||
for run in selected:
|
||||
load = run.get('load', {})
|
||||
ingest = load.get('ingest', {})
|
||||
if target == 'tempo' and 'ingest' not in load:
|
||||
ingest = load.get('transport', {}).get('ingest', {})
|
||||
health = run.get('post_load_health', {})
|
||||
execution = run.get('execution', {})
|
||||
start, finish = run.get('started_unix_ns'), run.get('finished_unix_ns')
|
||||
elapsed = (finish - start) / 1_000_000 if isinstance(start, int) and isinstance(finish, int) else None
|
||||
seconds = ingest.get('wall_seconds')
|
||||
rows.append([target, run.get('corpus'), run.get('load_summary', {}).get('after'),
|
||||
duration(elapsed), duration(seconds * 1000 if isinstance(seconds, (int, float)) else None),
|
||||
ingest.get('successful_requests'), ingest.get('failed_requests'),
|
||||
execution.get('load_workers_requested'), execution.get('max_in_flight'),
|
||||
execution.get('db_cpus'), execution.get('db_memory'),
|
||||
health.get('running'), health.get('oom_killed'),
|
||||
run.get('cleanup', {}).get('verified_absent')])
|
||||
return '\n'.join(['# Tracesbench load / lifecycle summary', '',
|
||||
'Total is the recorded target lifecycle; ingest is the sender wall time, '
|
||||
'not the entire load/readiness phase. Health is the post-load snapshot. '
|
||||
'Missing evidence remains blank.', '',
|
||||
table(['DB', 'Dataset', 'Loaded spans', 'Total', 'Ingest',
|
||||
'Write success', 'Write errors', 'Load workers', 'Max in flight',
|
||||
'DB CPU', 'DB memory', 'Running after load', 'OOM after load',
|
||||
'Container removed'], rows), '',
|
||||
render_resources([('dataset', root / 'runs' / 'generate')] +
|
||||
[(t, root / 'runs' / t) for t in targets], table)])
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument('--root', type=Path, required=True)
|
||||
parser.add_argument('--section', choices=('benchmark', 'lifecycle'), required=True)
|
||||
parser.add_argument('--targets', default=','.join(TARGETS), help='Comma-separated selected engines')
|
||||
args = parser.parse_args()
|
||||
targets = args.targets.split(',')
|
||||
if not targets or any(t not in TARGETS for t in targets) or len(set(targets)) != len(targets):
|
||||
parser.error('targets must be a non-empty comma-separated subset of greptimedb,victoriatraces,tempo')
|
||||
report, runs = load(args.root)
|
||||
print(benchmark(report, runs, targets) if args.section == 'benchmark' else lifecycle(report, runs, args.root, targets))
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
Reference in New Issue
Block a user