""" Performance benchmark: test different window_seconds and step combinations against real endpoints to find the optimal configuration for catch-up speed. Tests a representative mix of metrics (high-cardinality and low-cardinality) across all 17 endpoints, timing the full extract cycle. """ import json import time import requests import yaml import os import statistics RETRY_ATTEMPTS = 2 RETRY_BACKOFF = 2 # Use a recent timestamp that's guaranteed to have data # (Mar 17 2026 00:00 UTC -- pipeline has already extracted this range) BASE_START = 1773705600 def fetch_query_range(base_url, query, start, end, step): url = base_url.rstrip("/") + "/query_range" params = {"query": query, "start": start, "end": end, "step": step} for attempt in range(1, RETRY_ATTEMPTS + 1): try: resp = requests.post(url, data=params, timeout=300) resp.raise_for_status() data = resp.json() if data.get("status") != "success": return [], 0 results = data["data"]["result"] row_count = sum(len(s.get("values", [])) for s in results) return results, row_count except Exception as e: if attempt == RETRY_ATTEMPTS: return [], 0 time.sleep(RETRY_BACKOFF ** attempt) def load_config(): config_path = os.path.join(os.path.dirname(__file__), "..", "sources.yml") with open(config_path) as f: return yaml.safe_load(f) def run_test(endpoints, metric_name, metric_query, window_seconds, step): """Run one full extraction cycle: all endpoints for one metric, one window.""" start = BASE_START end = start + window_seconds total_rows = 0 total_series = 0 fetch_times = [] errors = 0 for ep in endpoints: t0 = time.time() results, rows = fetch_query_range(ep["url"], metric_query, start, end - 1, step) elapsed = time.time() - t0 fetch_times.append(elapsed) total_rows += rows total_series += len(results) if not results and rows == 0: errors += 1 return { "total_rows": total_rows, "total_series": total_series, "total_fetch_time": sum(fetch_times), "max_fetch_time": max(fetch_times) if fetch_times else 0, "avg_fetch_time": statistics.mean(fetch_times) if fetch_times else 0, "errors": errors, } def main(): config = load_config() endpoints = config["endpoints"] # Test metrics: pick a high-cardinality one and a low-cardinality one test_metrics = [ ("tikv_store_size_bytes", "tikv_store_size_bytes"), # high cardinality ("resource_manager_resource_unit_read_request_unit_sum", "resource_manager_resource_unit_read_request_unit_sum"), # medium ("serverless_metering_rcu_provisioned_capacity", "serverless_metering_rcu_provisioned_capacity"), # low cardinality ] # Configurations to test configs = [ # (window_seconds, step, label) (3600, 300, "1h window / 5min step (original)"), (3600, 1800, "1h window / 30min step"), (3600, 3600, "1h window / 1h step"), (7200, 1800, "2h window / 30min step (current)"), (7200, 3600, "2h window / 1h step"), (7200, 7200, "2h window / 2h step"), (14400, 1800, "4h window / 30min step"), (14400, 3600, "4h window / 1h step"), (14400, 7200, "4h window / 2h step"), (21600, 3600, "6h window / 1h step"), (21600, 7200, "6h window / 2h step"), (43200, 3600, "12h window / 1h step"), (43200, 7200, "12h window / 2h step"), (86400, 3600, "24h window / 1h step"), (86400, 7200, "24h window / 2h step"), ] results = [] print(f"{'Config':<35} {'Metric':<25} {'Window':>6} {'Step':>5} " f"{'Rows':>10} {'Series':>7} {'FetchT':>7} {'MaxT':>6} {'Err':>3}") print("-" * 120) for window_seconds, step, label in configs: for metric_name, metric_query in test_metrics: short_name = metric_name[:24] t0 = time.time() r = run_test(endpoints, metric_name, metric_query, window_seconds, step) wall_time = time.time() - t0 row = { "config": label, "metric": metric_name, "window_seconds": window_seconds, "step": step, "rows": r["total_rows"], "series": r["total_series"], "fetch_time_s": round(r["total_fetch_time"], 1), "max_fetch_s": round(r["max_fetch_time"], 1), "wall_time_s": round(wall_time, 1), "errors": r["errors"], "rows_per_sec": round(r["total_rows"] / max(wall_time, 0.1)), "hours_covered": window_seconds / 3600, } results.append(row) print(f"{label:<35} {short_name:<25} {window_seconds:>5}s {step:>5} " f"{r['total_rows']:>10,} {r['total_series']:>7,} " f"{r['total_fetch_time']:>6.1f}s {r['max_fetch_time']:>5.1f}s " f"{r['errors']:>3}") # Summary: estimate catch-up performance print("\n" + "=" * 120) print("CATCH-UP PERFORMANCE ESTIMATE") print("=" * 120) print(f"\n{'Config':<35} {'Hours/window':>12} {'All-metric time':>15} " f"{'Total rows':>12} {'Throughput':>14} {'Est catch-up':>14}") print("-" * 105) # Group by config config_groups = {} for r in results: key = r["config"] if key not in config_groups: config_groups[key] = {"rows": 0, "wall_time": 0, "hours_covered": r["hours_covered"], "window": r["window_seconds"], "step": r["step"]} config_groups[key]["rows"] += r["rows"] config_groups[key]["wall_time"] += r["wall_time_s"] # We tested 3 metrics; extrapolate to all 10 METRIC_SCALE = 10 / 3 for label, g in config_groups.items(): est_all_metrics_time = g["wall_time"] * METRIC_SCALE est_rows = int(g["rows"] * METRIC_SCALE) hours_per_window = g["hours_covered"] # How many windows fit in 25-min budget? if est_all_metrics_time > 0: windows_per_budget = max(1, int(25 * 60 / est_all_metrics_time)) else: windows_per_budget = 1 hours_per_run = hours_per_window * windows_per_budget # 2 runs per hour (every 30 min) hours_per_real_hour = hours_per_run * 2 # 100 hours to catch up catch_up_hours = 100 / max(hours_per_real_hour, 0.1) print(f"{label:<35} {hours_per_window:>10.0f}h " f"{est_all_metrics_time:>13.0f}s " f"{est_rows:>12,} " f"{hours_per_real_hour:>10.0f}h/h " f"{catch_up_hours:>10.1f}h") # Save raw results with open(os.path.join(os.path.dirname(__file__), "benchmark_results.json"), "w") as f: json.dump(results, f, indent=2) print("\nRaw results saved to benchmark_results.json") if __name__ == "__main__": main()