# 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()