ci: compare GreptimeDB File and S3 in observability benchmarks (#10187)

* ci: compare GreptimeDB File and S3 in observability benchmarks

Signed-off-by: WenyXu <wenymedia@gmail.com>

* ci: separate File and S3 benchmark steps and verify report artifacts

Signed-off-by: WenyXu <wenymedia@gmail.com>

---------

Signed-off-by: WenyXu <wenymedia@gmail.com>

.github/scripts/agent-observability-summary.py

@@ -133,9 +133,10 @@ def benchmark(root, targets):
     return '\n'.join(lines)
 
 
-def lifecycle(root, targets):
-    manifest = read(root / 'run-manifest.json')
-    corpus = read(root / 'corpus' / 'summary.json')
+def lifecycle(root, targets, data_root=None):
+    data_root = data_root or root
+    manifest = read(data_root / 'run-manifest.json')
+    corpus = read(data_root / 'corpus' / 'summary.json')
     lines = ['# Load / lifecycle summary', '',
              f'Dataset: **{cell(manifest.get("profile", corpus.get("data_profile", {}).get("id")))}**; rows: **{cell(corpus.get("line_count"))}**; '
              f'seed: `{cell(corpus.get("seed"))}`.',
@@ -161,12 +162,13 @@ def lifecycle(root, targets):
     physical_rows = []
     load_rows = []
     details = ['<details>', '<summary>DDL, index configuration and image provenance</summary>', '']
-    raw_bytes = read(root / 'corpus' / 'size.json').get('raw_jsonl_bytes')
+    raw_bytes = read(data_root / 'corpus' / 'size.json').get('raw_jsonl_bytes')
     for t in targets:
         directory = result_dir(root, t)
         physical = read(directory / 'physical-evidence.json')
         storage = physical.get('storage', {})
-        total = storage.get('engine_total_bytes')
+        s3 = read(root / f'{t}-1' / 'storage.json').get('type') == 's3'
+        total = None if s3 else storage.get('engine_total_bytes')
         stats = read(root / f'{t}-1' / 'post-load-docker-stats.json')
         ratio = f'{raw_bytes / total:.2f}×' if (physical.get('passed') is True and
                  isinstance(raw_bytes, (int, float)) and raw_bytes > 0 and
@@ -186,6 +188,7 @@ def lifecycle(root, targets):
     lines += ['## Load and physical evidence', '', table(['DB', 'Write transport', 'Batches', 'Load smoke rows'], load_rows), '',
               f'Raw JSONL size: **{size(raw_bytes)}**. Ratio = raw JSONL / engine bytes at collection time; '
               'not a settled-storage guarantee. CPU/memory are post-load snapshots, not peaks. '
+              'S3 total/compression is omitted: local files exclude remote objects. '
               'Missing artifacts remain blank; pre-cleanup stats are not substituted.', '',
               table(['DB', 'Engine size', 'Size ratio', 'Materialization', 'SSTs / active parts',
                      'Post-load CPU', 'Post-load memory / limit'], physical_rows), '', *details, '</details>', '']

.github/scripts/benchmark-version-comparison.py

@@ -62,11 +62,12 @@ def row(query, metric, old, new, compatible, reason):
                 improvement_percent=delta, status='comparable' if delta is not None else reason)
 
 
-def agent(root):
+def agent(root, storage="file"):
+    candidate, baseline = (root, root / "baseline") if storage == "file" else (root / "s3", root / "baseline/s3")
     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)]
+    a, b = load(baseline), load(candidate)
+    finished = [r / 'greptimedb-1/exit-code.txt' for r in (baseline, candidate)]
     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')))
@@ -81,8 +82,9 @@ def agent(root):
     return rows
 
 
-def traces(root):
-    a, b = (read(r / 'runs/greptimedb/run.json') for r in (root / 'baseline', root))
+def traces(root, storage="file"):
+    candidate, baseline = (root, root / "baseline") if storage == "file" else (root / "s3", root / "baseline/s3")
+    a, b = (read(r / 'runs/greptimedb/run.json') for r in (baseline, candidate))
     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', {}),
@@ -110,18 +112,22 @@ def traces(root):
     return rows
 
 
-def long_range(root):
+def long_range(root, storage="file"):
+    names = ("greptimedb-baseline", "greptimedb") if storage == "file" else ("greptimedb-s3-baseline", "greptimedb-s3")
+    def target_root(name):
+        nested = root / "o11ybench" / ("o11ybench-" + name)
+        return nested if nested.exists() else root
     rows = []
-    manifest = read(root / 'run-manifest.json')
+    manifests = {name: read(target_root(name) / 'run-manifest.json') for name in names}
     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()
+        all(field in (manifests[target].get(phase, {}).get(target, {}).get('docker', {}) if phase == 'post_load_results'
+                      else manifests[target].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 target in names)
     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'))
+        a, b = (read(target_root(name) / name / 'results-vmbench-long-range' / (mode + '-summary.json'))
+                for name in names)
         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')
@@ -178,7 +184,13 @@ def main():
         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)}
+                'rows': []}
+    mode = os.environ.get('GREPTIMEDB_STORAGE', 'file')
+    for storage in (['file', 's3'] if mode == 'both' else [mode]):
+        rows = {'agent': agent, 'traces': traces, 'long-range': long_range}[args.kind](args.root, storage)
+        for item in rows:
+            item['query'] = storage.upper() + '/' + item['query']
+        document['rows'].extend(rows)
     text = render(document)
     # Preserve partial comparisons even if a target failed before creating output.
     args.root.mkdir(parents=True, exist_ok=True)

.github/scripts/o11ybench-storage.py

@@ -0,0 +1,295 @@
+# 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.
+
+"""Select File/S3 benchmark arms without changing workload adapters."""
+import argparse
+import copy
+import importlib.util
+import json
+from pathlib import Path
+import sys
+import os
+import subprocess
+
+
+def validate(env):
+    mode = env.get('GREPTIMEDB_STORAGE', 'file')
+    if mode not in ('file', 's3', 'both'):
+        raise ValueError('GREPTIMEDB_STORAGE must be file, s3 or both')
+    if mode != 'file':
+        if env.get('RUN_GREPTIMEDB') != 'true':
+            raise ValueError('S3 requires Run GreptimeDB')
+        if env.get('PROVIDER') != 'AWS':
+            raise ValueError('S3 comparisons require an AWS runner')
+        if not env.get('S3_BUCKET') or not env.get('S3_REGION'):
+            raise ValueError('AWS_CI_TEST_BUCKET and AWS_CI_TEST_BUCKET_REGION must be configured')
+        if env['S3_REGION'] != env.get('RUNNER_REGION'):
+            raise ValueError('S3 bucket and EC2 runner regions must match')
+    return mode
+
+
+def replace(command, key, value):
+    result = list(command)
+    if key in result:
+        result[result.index(key) + 1] = value
+    else:
+        result += [key, value]
+    return result
+
+
+def commands(command, mode):
+    """Recovery visits each owned root; execution uses explicit workflow steps."""
+    stage = command[command.index('--stage') + 1]
+    if stage != 'cleanup':
+        raise ValueError('only cleanup may expand multiple arms')
+    result = [list(command)]
+    if mode != 'file':
+        root = command[command.index('--root') + 1]
+        owner = command[command.index('--owner') + 1]
+        secondary = replace(command, '--root', root + '/s3')
+        result.append(replace(secondary, '--owner', owner + '-s3'))
+    return result
+
+
+def module(path):
+    sys.path.insert(0, str(path.parent))
+    spec = importlib.util.spec_from_file_location(path.stem.replace('-', '_'), path)
+    result = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(result)
+    return result
+
+
+def summary(kind, root, targets, section, mode):
+    filename = 'agent-observability-summary.py' if kind == 'agent' else 'tracesbench-summary.py'
+    renderer = module(Path(__file__).with_name(filename))
+    primary = [t for t in targets if t != 'greptimedb' or mode != 's3']
+    arms = [(root, primary, 'File')]
+    if mode != 'file' and 'greptimedb' in targets:
+        arms.append((root / 's3', ['greptimedb'], 'S3'))
+    output = []
+    for arm, selected, label in arms:
+        if not selected:
+            continue
+        output.append('## ' + ('GreptimeDB S3' if label == 'S3' else 'Primary targets (GreptimeDB File)'))
+        if kind == 'agent':
+            body = (renderer.benchmark(arm, selected) if section == 'benchmark'
+                    else renderer.lifecycle(arm, selected, data_root=root.parent if root.name == 'baseline' else root))
+        else:
+            report, runs = renderer.load(arm)
+            body = (renderer.benchmark(report, runs, selected) if section == 'benchmark'
+                    else renderer.lifecycle(report, runs, arm, selected))
+        body = body.replace('| greptimedb |', '| GreptimeDB ' + label + ' |')
+        body = body.replace('**greptimedb**', '**GreptimeDB ' + label + '**')
+        output.append(body)
+        if label == 'S3':
+            directories = [arm / 'greptimedb-1'] if kind == 'agent' else [arm / 'runs/greptimedb']
+            for directory in directories:
+                for action in ('probe', 'inventory', 'cleanup'):
+                    path = directory / ('s3-' + action + '.json')
+                    data = json.loads(path.read_text()) if path.exists() else {}
+                    status = 'passed' if data.get('passed') is True else 'failed / unavailable'
+                    count = data.get('object_count', data.get('deleted_objects', '—'))
+                    output.append(f'- S3 {action}: **{status}**; objects: {count}.')
+    return '\n\n'.join(output)
+
+
+def split_metrics(plan):
+    """One executor root per target, with the immutable dataset shared by all."""
+    result = {}
+    for target in plan['targets']:
+        child = copy.deepcopy(plan)
+        name = target['name']
+        dest = str(Path(plan['output_root']) / 'o11ybench' / ('o11ybench-' + name))
+        child['output_root'] = dest
+        child['run_id'] = plan['run_id'] + '-' + name
+        child['targets'] = [copy.deepcopy(target)]
+        child['data']['generate'] = False
+        child['report'].update(root=dest + '/report', index_html=dest + '/report/index.html')
+        child['artifacts'] = {k: dest + '/' + Path(v).name for k, v in plan['artifacts'].items()}
+        result[name] = child
+    return result
+
+
+def metrics(action, args):
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--plan', type=Path, required=True)
+    parser.add_argument('--tools', type=Path, required=True)
+    parser.add_argument('--runtime-image', required=True)
+    parser.add_argument('--target')
+    a = parser.parse_args(args)
+    plan = json.loads(a.plan.read_text())
+    plan['runtime_image'] = a.runtime_image
+    executor = module(a.tools / 'scripts/run_benchmark_plan.py')
+    root = Path(plan['output_root'])
+    if action == 'metrics-prepare':
+        executor.require_safe_execute_root(root)
+        root.mkdir(parents=True, exist_ok=False)
+        manifest = executor.build_manifest(plan)
+        executor.run_checked([str(executor.DOWNLOAD_DATA)], manifest,
+                             env=executor.generation_environment(plan))
+        (root / 'generation.json').write_text(json.dumps(manifest, indent=2) + '\n')
+        for name, child in split_metrics(plan).items():
+            (a.plan.parent / (name + '.json')).write_text(json.dumps(child, indent=2) + '\n')
+    elif action == 'metrics-run':
+        path = a.plan.parent / (a.target + '.json')
+        child = json.loads(path.read_text())
+        executor.execute_plan(child, executor.build_manifest(child), replace=False)
+        subprocess.run([sys.executable, str(a.tools / 'scripts/check_run_benchmark_plan.py'),
+                        '--artifacts', child['output_root']], check=True)
+    else:
+        # Validate every selected target together, retaining cross-target gates.
+        manifests = [executor.build_manifest(child) for child in split_metrics(plan).values()]
+        combined = executor.build_manifest(plan)
+        combined['targets'] = [target for m in manifests for target in m['targets']]
+        for command in executor.validation_commands(combined):
+            subprocess.run(command, check=True)
+
+
+def overview(path):
+    """Render the same validated report data into the GitHub job summary."""
+    data = json.loads(path.read_text())
+    renderer = module(Path(__file__).with_name('agent-observability-summary.py'))
+    lines = ['# Storage comparison', '', 'Smoke results are diagnostic; see classification per arm. '
+             'No cold-object-store performance claim. Missing results are not passes.', '']
+    for workload in data['workloads']:
+        labels = {p['id']: p['label'] for p in workload['participants']}
+        for scenario in workload['scenarios']:
+            lines += ['## ' + workload['title'] + ' / ' + scenario['label'], '']
+            rows = []
+            for result in scenario['results']:
+                lines += [labels[result['participant']] + ': ' + str(result.get('classification')), '']
+                for query_id, query in result['queries'].items():
+                    m = query['metrics']
+                    rows.append([query_id, labels[result['participant']], query['status'],
+                                 renderer.duration(m.get('p50_ms', m.get('median_ms'))),
+                                 renderer.duration(m.get('p95_ms')), renderer.duration(m.get('p99_ms'))])
+            lines.append(renderer.table(['Query', 'DB / storage', 'Status', 'P50 / median', 'P95', 'P99'], rows))
+    return '\n'.join(lines)
+
+
+def audit(root, report, targets, mode, baseline=False):
+    data = json.loads(report.read_text())
+    actual = {(p['target'], p.get('storage', {}).get('type'), p['role'])
+              for w in data['workloads'] for p in w['participants']}
+    expected = {(t, None, 'comparison') for t in targets if t != 'greptimedb'}
+    modes = ['file', 's3'] if mode == 'both' else [mode]
+    if 'greptimedb' in targets:
+        expected |= {('greptimedb', m, role) for m in modes
+                     for role in (['candidate', 'baseline'] if baseline else ['candidate'])}
+    if actual != expected:
+        raise ValueError(f'participant mismatch: expected {sorted(map(str, expected))}, got {sorted(map(str, actual))}')
+    for w in data['workloads']:
+        for scenario in w['scenarios']:
+            for result in scenario['results']:
+                queries = result['queries']
+                if not any(q['status'] == 'passed' for q in queries.values()):
+                    raise ValueError('no successful measured queries: ' + result['participant'])
+                if any(q['status'] in ('failed', 'unavailable') for q in queries.values()):
+                    raise ValueError('failed or unavailable queries: ' + result['participant'])
+    for manifest_path in root.glob('o11ybench/*/run-manifest.json'):
+        manifest = json.loads(manifest_path.read_text())
+        if manifest.get('status') != 'passed':
+            raise ValueError('incomplete target manifest: ' + str(manifest_path))
+    probes = list(root.rglob('s3-probe.json'))
+    expected_s3 = sum(m == 's3' for _, m, _ in expected)
+    if len(probes) != expected_s3:
+        raise ValueError('missing or duplicate S3 evidence')
+    for probe in probes:
+        for action in ('probe', 'inventory', 'cleanup'):
+            payload = json.loads(probe.with_name('s3-' + action + '.json').read_text())
+            if payload.get('passed') is not True or (action == 'inventory' and payload.get('object_count', 0) <= 0):
+                raise ValueError('failed S3 ' + action)
+    text = (root / 'ci-summary.md').read_text()
+    for m in modes if 'greptimedb' in targets else []:
+        if m.upper() not in text:
+            raise ValueError('storage mode missing from saved summary: ' + m)
+    result = dict(passed=True, participants=[dict(target=t, storage=m, role=role)
+                  for t, m, role in sorted(expected, key=str)], s3_arms=len(probes))
+    (root / 'artifact-audit.json').write_text(json.dumps(result, indent=2) + '\n')
+    lines = ['## Artifact validation', '', '- Selected participants: ' + str(len(expected)) + '; all present.',
+             '- Query result coverage and S3 probe/inventory/cleanup: passed.', '']
+    renderer = module(Path(__file__).with_name('agent-observability-summary.py'))
+    rows = []
+    for probe in probes:
+        inv = json.loads(probe.with_name('s3-inventory.json').read_text())
+        cleanup = json.loads(probe.with_name('s3-cleanup.json').read_text())
+        rows.append([str(probe.parent.relative_to(root)), inv['object_count'], inv['object_bytes'],
+                     cleanup.get('deleted_objects'), 'passed'])
+    if rows:
+        lines.append(renderer.table(['S3 arm', 'Objects', 'Remote bytes', 'Deleted current objects', 'Cleanup'], rows))
+    text = '\n'.join(lines) + '\n'
+    with (root / 'ci-summary.md').open('a') as output:
+        output.write(text)
+    if os.environ.get('GITHUB_STEP_SUMMARY'):
+        with Path(os.environ['GITHUB_STEP_SUMMARY']).open('a') as output:
+            output.write(text)
+    return result
+
+
+def main():
+    if len(sys.argv) > 1 and sys.argv[1] == 'audit':
+        parser = argparse.ArgumentParser()
+        parser.add_argument('--root', type=Path, required=True)
+        parser.add_argument('--report', type=Path, required=True)
+        parser.add_argument('--targets', required=True)
+        args = parser.parse_args(sys.argv[2:])
+        print(json.dumps(audit(args.root, args.report, args.targets.split(','),
+                              os.environ.get('GREPTIMEDB_STORAGE', 'file'),
+                              os.environ.get('COMPARE_GREPTIMEDB') == 'true')))
+        return
+    if len(sys.argv) > 1 and sys.argv[1] == 'normalize-metrics':
+        path = Path(sys.argv[2])
+        plan = json.loads(path.read_text())
+        for target in plan['targets']:
+            if target.get('storage', {}).get('type') == 's3' and '-s3' not in target['name']:
+                target['name'] = target['name'].replace('-baseline', '-s3-baseline') if target['name'].endswith('-baseline') else target['name'] + '-s3'
+        path.write_text(json.dumps(plan, indent=2) + '\n')
+        return
+    if len(sys.argv) > 1 and sys.argv[1].startswith('metrics-'):
+        metrics(sys.argv[1], sys.argv[2:])
+        return
+    if len(sys.argv) > 1 and sys.argv[1] == 'overview':
+        print(overview(Path(sys.argv[2])))
+        return
+    if len(sys.argv) > 1 and sys.argv[1] == 'summary':
+        parser = argparse.ArgumentParser()
+        parser.add_argument('--kind', choices=['agent', 'traces'], required=True)
+        parser.add_argument('--root', type=Path, required=True)
+        parser.add_argument('--targets', required=True)
+        parser.add_argument('--section', choices=['benchmark', 'lifecycle'], required=True)
+        args = parser.parse_args(sys.argv[2:])
+        print(summary(args.kind, args.root, args.targets.split(','), args.section,
+                      os.environ.get('GREPTIMEDB_STORAGE', 'file')))
+        return
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('action', choices=['validate', 'run'])
+    parser.add_argument('command', nargs=argparse.REMAINDER)
+    args = parser.parse_args()
+    mode = validate(os.environ)
+    if args.action == 'validate':
+        print('GreptimeDB storage: ' + mode)
+        return
+    command = args.command[1:] if args.command[:1] == ['--'] else args.command
+    planned = commands(command, mode)
+    failed = 0
+    for cmd in planned:
+        result = subprocess.run(cmd)
+        failed = failed or result.returncode
+        # Cleanup must try both arms even if one failed. Target failures remain
+        # visible, while the other arm can still produce diagnostic evidence.
+    raise SystemExit(failed)
+
+
+if __name__ == '__main__':
+    main()