diff --git a/evalbench_service/supervisord_cloudrun.conf b/evalbench_service/supervisord_cloudrun.conf index 83fae017..33a80acc 100644 --- a/evalbench_service/supervisord_cloudrun.conf +++ b/evalbench_service/supervisord_cloudrun.conf @@ -19,6 +19,8 @@ command=python /evalbench/viewer/run_precompute.py directory=/evalbench/viewer autostart=true autorestart=true + +environment=PRECOMPUTE_WORKERS=12,PRECOMPUTE_INTERVAL=600 stdout_logfile=/dev/stdout stdout_logfile_maxbytes=0 stderr_logfile=/dev/stderr diff --git a/viewer/main.py b/viewer/main.py index 038c8c7c..c1ee4f2a 100644 --- a/viewer/main.py +++ b/viewer/main.py @@ -7,6 +7,7 @@ import subprocess import precompute_trends import dataset_quality +from run_index import list_run_directories from summarizer import summarize_eval_scoring from ai_comparer import compare_evals @@ -229,14 +230,7 @@ def get_color_for_pct(val_str): def on_load(e: me.LoadEvent): state = me.state(State) results_dir = get_results_dir() - directories = [] - if os.path.exists(results_dir): - # List directories only - directories = [ - d - for d in os.listdir(results_dir) - if os.path.isdir(os.path.join(results_dir, d)) - ] + directories = list_run_directories(results_dir) job_id = me.query_params.get("job_id") or me.query_params.get("jobid") if job_id and job_id in directories: @@ -259,14 +253,8 @@ def on_load(e: me.LoadEvent): def status_component(): results_dir = get_results_dir() - directories = [] - if os.path.exists(results_dir): - directories = [ - d - for d in os.listdir(results_dir) - if os.path.isdir(os.path.join(results_dir, d)) - ] - + directories = list_run_directories(results_dir) + with me.box( style=me.Style( background="#ffffff", @@ -2055,15 +2043,8 @@ def render_app_content(): results_dir = get_results_dir() logging.info(f"render_app_content: selected_directory='{state.selected_directory}', selected_evals='{state.selected_evals}', selected_main_tab='{state.selected_main_tab}'") - directories = [] - if os.path.exists(results_dir): - # List directories only - directories = [ - d - for d in os.listdir(results_dir) - if os.path.isdir(os.path.join(results_dir, d)) - ] - + directories = list_run_directories(results_dir) + def on_title_click(e: me.ClickEvent): state.selected_directory = "" state.conversation_index = 0 @@ -2084,7 +2065,9 @@ def on_clear_cache_click(e: me.ClickEvent): os.remove(filters_cache_file) logging.info("Cleared precomputed files. Triggering precompute...") - + + run_index.invalidate() + import threading threading.Thread(target=precompute_trends.precompute).start() @@ -2289,12 +2272,22 @@ def get_val(cfg_name): me.text("AI Summary", type="headline-5") if not state.ai_summary and state.selected_directory: + # The cache only carries summaries for the recent window. + state.ai_summary = precompute_trends.read_ai_summary( + results_dir, state.selected_directory + ) + trends_cache_file = os.path.join(results_dir, "trends_cache.csv") if os.path.exists(trends_cache_file): try: - cache_df = pd.read_csv(trends_cache_file) + wanted = ['job_id', 'ai_score'] + if not state.ai_summary: + wanted.append('ai_summary') + cache_df = pd.read_csv( + trends_cache_file, usecols=lambda c: c in wanted + ) run_data = cache_df[cache_df['job_id'] == state.selected_directory] - if not run_data.empty and 'ai_summary' in run_data.columns: + if not state.ai_summary and not run_data.empty and 'ai_summary' in run_data.columns: summary = run_data['ai_summary'].values[0] if not pd.isna(summary) and summary != "N/A": state.ai_summary = summary diff --git a/viewer/precompute_dataset_quality.py b/viewer/precompute_dataset_quality.py index e623860e..67b62adc 100644 --- a/viewer/precompute_dataset_quality.py +++ b/viewer/precompute_dataset_quality.py @@ -13,6 +13,8 @@ import logging import os +from run_index import list_run_directories + logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') CACHE_FILENAME = "dataset_quality_cache.json" @@ -146,8 +148,8 @@ def precompute(): logging.warning("Could not read %s, rebuilding: %s", processed_dirs_file, e) directories = [ - d for d in os.listdir(results_dir) - if os.path.isdir(os.path.join(results_dir, d)) and d not in processed + d for d in list_run_directories(results_dir, force=True) + if d not in processed ] logging.info("Dataset quality: %d new directories to scan", len(directories)) diff --git a/viewer/precompute_lease.py b/viewer/precompute_lease.py new file mode 100644 index 00000000..6594f011 --- /dev/null +++ b/viewer/precompute_lease.py @@ -0,0 +1,140 @@ +"""Single-writer lease so only one instance precomputes at a time. + +Supervisord starts a precompute loop in every Cloud Run instance, but they share +one results directory and the caches are rewritten whole, so a later write +silently discards an earlier one's rows while processed_dirs.json still records +those runs as done -- they are then never reconsidered. + +Acquiring writes a holder id and timestamp, waits out the window in which a +competing write could still land, then reads back: a whole-file rewrite has one +winner and GCS reads are strongly consistent after it, so whoever reads its own +id back is the sole holder. Timestamps are wall-clock because they are compared +across hosts, where monotonic clocks are meaningless. +""" + +import logging +import os +import socket +import threading +import time +import uuid + +LEASE_FILENAME = "precompute.lease" + +# Long enough to ride out a slow FUSE write, short enough that a killed holder +# does not stall precompute for long. +LEASE_TTL_SECONDS = float(os.environ.get("PRECOMPUTE_LEASE_TTL", "120")) +RENEW_INTERVAL_SECONDS = LEASE_TTL_SECONDS / 4 + +# Read-back is only meaningful once any competing write has had time to land. +GUARD_SECONDS = float(os.environ.get("PRECOMPUTE_LEASE_GUARD", "5")) + +_HOLDER_ID = f"{socket.gethostname()}:{os.getpid()}:{uuid.uuid4().hex[:8]}" + + +def _lease_path(results_dir): + return os.path.join(results_dir, LEASE_FILENAME) + + +def _read(results_dir): + """Return (holder, timestamp), or (None, 0.0) if absent or unreadable.""" + try: + with open(_lease_path(results_dir)) as f: + holder, _, timestamp = f.read().partition("\n") + return holder.strip(), float(timestamp.strip()) + except (OSError, ValueError): + # A torn read lands here too; safe, because read-back must still pass. + return None, 0.0 + + +def _write(results_dir): + with open(_lease_path(results_dir), "w") as f: + f.write(f"{_HOLDER_ID}\n{time.time()}\n") + + +class PrecomputeLease: + """Context manager guarding a precompute pass. Truthy only if held.""" + + def __init__(self, results_dir): + self.results_dir = results_dir + self.acquired = False + self._stop = threading.Event() + self._renewer = None + + def acquire(self): + holder, timestamp = _read(self.results_dir) + age = time.time() - timestamp + + if holder and holder != _HOLDER_ID and age < LEASE_TTL_SECONDS: + logging.info( + "Precompute lease held by %s (renewed %.0fs ago); skipping pass", + holder, age, + ) + return False + if holder and holder != _HOLDER_ID: + logging.warning( + "Taking over precompute lease from %s, stale for %.0fs", holder, age + ) + + try: + _write(self.results_dir) + except OSError: + # Fail open: a dashboard that silently stops updating is worse than + # the duplicated work this lease exists to prevent. + logging.exception("Could not write precompute lease; running unguarded") + self.acquired = True + return True + + time.sleep(GUARD_SECONDS) + + winner, _ = _read(self.results_dir) + if winner != _HOLDER_ID: + logging.info("Lost precompute lease race to %s; skipping pass", winner) + return False + + self.acquired = True + self._renewer = threading.Thread( + target=self._renew_loop, name="precompute-lease-renew", daemon=True + ) + self._renewer.start() + logging.info("Acquired precompute lease as %s", _HOLDER_ID) + return True + + def _renew_loop(self): + while not self._stop.wait(RENEW_INTERVAL_SECONDS): + holder, _ = _read(self.results_dir) + if holder and holder != _HOLDER_ID: + # Stand down rather than fight the successor for the file. + logging.warning("Precompute lease taken over by %s; stopping renewal", holder) + return + try: + _write(self.results_dir) + except OSError: + logging.warning("Could not renew precompute lease") + + def release(self): + self._stop.set() + if self._renewer: + self._renewer.join(timeout=RENEW_INTERVAL_SECONDS) + self._renewer = None + if not self.acquired: + return + self.acquired = False + # Only clear the file if it is still ours, never a successor's. + holder, _ = _read(self.results_dir) + if holder == _HOLDER_ID: + try: + os.remove(_lease_path(self.results_dir)) + except OSError: + logging.warning("Could not remove precompute lease file") + + def __enter__(self): + self.acquire() + return self + + def __exit__(self, *exc): + self.release() + return False + + def __bool__(self): + return self.acquired diff --git a/viewer/precompute_trends.py b/viewer/precompute_trends.py index f5571060..0d175e20 100644 --- a/viewer/precompute_trends.py +++ b/viewer/precompute_trends.py @@ -1,9 +1,14 @@ import os +import csv import logging import json import argparse import pandas as pd +from run_index import list_run_directories + +csv.field_size_limit(10**9) + logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') BATCH_SIZE = int(os.environ.get("PRECOMPUTE_BATCH_SIZE", 50)) @@ -14,6 +19,45 @@ # where the summarizer starts seeing 429s, so raising it further buys nothing. MAX_WORKERS = int(os.environ.get("PRECOMPUTE_WORKERS", 50)) +# ai_summary is ~96% of trends_cache.csv, which is read in full on every render. +# Only the recent window stays inline; every summary is also written beside its +# own run so nothing is lost. +AI_SUMMARY_FILENAME = "ai_summary.txt" +AI_SUMMARY_CACHE_DAYS = int(os.environ.get("AI_SUMMARY_CACHE_DAYS", 7)) + + +def ai_summary_path(results_dir, job_id): + return os.path.join(results_dir, job_id, AI_SUMMARY_FILENAME) + + +def write_ai_summary(run_dir, summary): + """Persist a summary beside its run so trimming the cache is not lossy.""" + if not isinstance(summary, str) or not summary or summary == "N/A": + return + try: + with open(os.path.join(run_dir, AI_SUMMARY_FILENAME), "w") as f: + f.write(summary) + except Exception: + logging.warning("Could not write %s for %s", AI_SUMMARY_FILENAME, run_dir) + + +def read_ai_summary(results_dir, job_id): + """Return the stored summary for a run, or "" if it has none.""" + try: + with open(ai_summary_path(results_dir, job_id)) as f: + return f.read() + except OSError: + return "" + + +def _is_recent(run_time, days=AI_SUMMARY_CACHE_DAYS): + """Whether a run is inside the window that keeps its summary inline.""" + ts = pd.to_datetime(run_time, errors='coerce') + if pd.isna(ts): + # Cannot be aged out safely, so treat as old and let the sidecar serve it. + return False + return ts >= pd.Timestamp.now() - pd.Timedelta(days=days) + def get_results_dir(): # Try to read from environment variable @@ -35,6 +79,20 @@ def get_results_dir(): return results_dir_candidates[1] # Fallback to default +class UnparsableRun(Exception): + """A run whose CSVs are present but structurally unusable. + + Distinct from "not ready yet" (no summary.csv), which returns None so a + later pass retries. A missing column looks the same on every pass, so such + a run is recorded as processed rather than retried forever. + """ + + def __init__(self, job_id, reason): + super().__init__(reason) + self.job_id = job_id + self.reason = reason + + def process_directory(d, results_dir): run_dir = os.path.join(results_dir, d) configs_file = os.path.join(run_dir, "configs.csv") @@ -47,6 +105,8 @@ def process_directory(d, results_dir): try: # Read configs configs_df = pd.read_csv(configs_file) + if 'config' not in configs_df.columns: + raise UnparsableRun(d, "configs.csv has no 'config' column") # Extract requester, product, dataset and generator requester_row = configs_df[configs_df['config'].str.contains('guitar_requester', na=False)] @@ -61,6 +121,8 @@ def process_directory(d, results_dir): # Read summary summary_df = pd.read_csv(summary_file) + if 'metric_name' not in summary_df.columns: + raise UnparsableRun(d, "summary.csv has no 'metric_name' column") # Extract metrics latency_row = summary_df[summary_df['metric_name'] == 'end_to_end_latency'] @@ -118,7 +180,11 @@ def get_metric_pct(row): except Exception as e: logging.warning(f"Error reading {os.path.basename(file_to_read)} for {d}: {e}") - run_time = summary_df['run_time'].values[0] if not summary_df.empty else "unknown" + run_time = ( + summary_df['run_time'].values[0] + if 'run_time' in summary_df.columns and not summary_df.empty + else "unknown" + ) if run_time != "unknown": try: run_time = pd.to_datetime(run_time).strftime('%Y-%m-%d %H:%M:%S') @@ -140,6 +206,8 @@ def get_metric_pct(row): except Exception as e: logging.error(f"Error generating AI summary for {d}: {e}") + write_ai_summary(run_dir, ai_summary) + logging.info(f"Successfully processed directory: {d}") return { 'run_time': run_time, @@ -157,11 +225,20 @@ def get_metric_pct(row): 'goal_completion': goal_completion, 'job_id': d, 'ai_score': ai_score, - 'ai_summary': ai_summary + 'ai_summary': ai_summary if _is_recent(run_time) else "", } + except UnparsableRun: + # Must precede the catch-all: returning None would requeue the run. + raise + except (pd.errors.ParserError, pd.errors.EmptyDataError, OSError) as e: + # Mid-write or a transient FUSE read; may well parse on a later pass. + logging.warning(f"Deferring {d}, artifacts not readable yet: {e}") + return None except Exception as e: + # Both files parsed, so the fault is in their shape and will recur on + # every pass. Requeueing it is what jammed the backlog. logging.exception(f"Error reading data from {d}") - return None + raise UnparsableRun(d, f"{type(e).__name__}: {e}") from e def _read_json(path, default): @@ -220,6 +297,75 @@ def _append_rows(cache_file, rows): new_df[header].to_csv(cache_file, mode="a", header=False, index=False) +class _NothingToTrim(Exception): + """The cache has no usable ai_summary column; leave it untouched.""" + + +def _discard(path): + try: + os.remove(path) + except OSError as e: + logging.debug("Best-effort discard failed for %s: %s", path, e) + + +def _trim_stale_summaries(results_dir, cache_file): + """Blank ai_summary on cache rows that have aged out of the inline window. + + Rows written by an earlier pass were recent when appended and go stale on + their own, so this runs every pass rather than only at write time. Each + summary is written to its sidecar first where one is missing -- without + that backfill this would destroy every summary already in the cache. + + Streamed a row at a time rather than loaded into a frame: the cache this + exists to shrink is exactly the one too large to hold in memory. The file + is only replaced if something was actually trimmed, so a cache already + inside the window costs one sequential read and no write. + """ + if not os.path.exists(cache_file) or os.path.getsize(cache_file) == 0: + return + + required = {'ai_summary', 'run_time', 'job_id'} + temp_file = cache_file + ".trim" + trimmed = 0 + backfilled = 0 + + try: + with open(cache_file, newline="") as src, open(temp_file, "w", newline="") as dst: + reader = csv.DictReader(src) + if not reader.fieldnames or not required <= set(reader.fieldnames): + raise _NothingToTrim + writer = csv.DictWriter(dst, fieldnames=reader.fieldnames) + writer.writeheader() + for row in reader: + if row.get('ai_summary') and not _is_recent(row.get('run_time')): + job_id = row.get('job_id') or "" + if job_id and not os.path.exists(ai_summary_path(results_dir, job_id)): + write_ai_summary( + os.path.join(results_dir, job_id), row['ai_summary'] + ) + backfilled += 1 + row['ai_summary'] = "" + trimmed += 1 + writer.writerow(row) + except _NothingToTrim: + _discard(temp_file) + return + except Exception: + logging.exception("Could not trim summaries from %s", cache_file) + _discard(temp_file) + return + + if not trimmed: + _discard(temp_file) + return + + os.replace(temp_file, cache_file) + logging.info( + "Trimmed %d aged-out summaries from the trends cache (%d sidecars backfilled)", + trimmed, backfilled, + ) + + def precompute(): results_dir = get_results_dir() logging.info(f"Reading results from {results_dir}") @@ -235,11 +381,7 @@ def precompute(): processed_dirs = set(_read_json(processed_dirs_file, [])) logging.info(f"Loaded {len(processed_dirs)} processed directories from state.") - all_directories = [ - d - for d in os.listdir(results_dir) - if os.path.isdir(os.path.join(results_dir, d)) - ] + all_directories = list_run_directories(results_dir, force=True) # Filter for new directories new_directories = [d for d in all_directories if d not in processed_dirs] @@ -249,6 +391,8 @@ def precompute(): if total_new == 0: logging.info("No new directories to process.") + # Rows still age out of the inline window with no new runs at all. + _trim_stale_summaries(results_dir, cache_file) return # Carried forward from the last pass rather than recomputed off the trends @@ -279,6 +423,10 @@ def precompute(): for future, directory in futures.items(): try: res = future.result() + except UnparsableRun as e: + logging.warning(f"Skipping {directory}: {e.reason}") + batch_processed.append(e.job_id) + continue except Exception: # One unreadable run must not cost the rest of the batch. logging.exception(f"Error processing {directory}") @@ -327,6 +475,8 @@ def precompute(): f"appended {len(rows)} rows, {len(processed_dirs)} directories processed." ) + _trim_stale_summaries(results_dir, cache_file) + logging.info(f"Precomputed trends data saved to {cache_file}") logging.info(f"Precomputed filter values saved to {filters_file}") logging.info(f"Saved {len(processed_dirs)} processed directories to state.") diff --git a/viewer/run_index.py b/viewer/run_index.py new file mode 100644 index 00000000..02067662 --- /dev/null +++ b/viewer/run_index.py @@ -0,0 +1,65 @@ +"""Cached listing of run directories under the results mount. + +The mount is GCS FUSE holding tens of thousands of run directories, so +os.listdir + os.path.isdir costs a stat round trip per entry. os.scandir +carries the entry type through from the listing and avoids them; the TTL +collapses the repeated walks the render path makes. +""" + +import logging +import os +import threading +import time + +CACHE_TTL_SECONDS = float(os.environ.get("RUN_DIRS_CACHE_TTL", "60")) + +_lock = threading.Lock() +_cache = {"path": None, "dirs": [], "at": 0.0} + + +def list_run_directories(results_dir, force=False): + """Return the names of run directories directly under results_dir. + + Cached for CACHE_TTL_SECONDS. On a failed refresh a previous listing for the + same path is returned rather than an empty one, so a transient FUSE error + does not blank the dashboard. + """ + now = time.monotonic() + + with _lock: + if ( + not force + and _cache["path"] == results_dir + and now - _cache["at"] < CACHE_TTL_SECONDS + ): + return list(_cache["dirs"]) + + try: + dirs = [] + with os.scandir(results_dir) as entries: + for entry in entries: + try: + if entry.is_dir(follow_symlinks=False): + dirs.append(entry.name) + except OSError: + # Skipping one bad entry beats failing the whole listing. + continue + except FileNotFoundError: + dirs = [] + except OSError: + logging.exception("Failed to list run directories in %s", results_dir) + with _lock: + if _cache["path"] == results_dir: + return list(_cache["dirs"]) + return [] + + with _lock: + _cache.update({"path": results_dir, "dirs": dirs, "at": now}) + + return list(dirs) + + +def invalidate(): + """Drop the cached listing so the next call re-walks the mount.""" + with _lock: + _cache.update({"path": None, "dirs": [], "at": 0.0}) diff --git a/viewer/run_precompute.py b/viewer/run_precompute.py index 91a805f5..a99897f1 100644 --- a/viewer/run_precompute.py +++ b/viewer/run_precompute.py @@ -7,22 +7,27 @@ import precompute_trends import precompute_dataset_quality +from precompute_lease import PrecomputeLease def main(): interval = int(os.environ.get("PRECOMPUTE_INTERVAL", 300)) + results_dir = precompute_trends.get_results_dir() while True: - # Dataset quality runs first because it is cheap and always finishes. The - # trends pass makes an LLM call per unprocessed run and gets SIGKILLed - # mid-pass on a large backlog, which the try/except below cannot catch and - # which would otherwise starve every pass after it. - for precompute in ( - precompute_dataset_quality.precompute, - precompute_trends.precompute, - ): - try: - precompute() - except Exception as e: - print(f"Error in {precompute.__module__}: {e}") + # Every instance runs this loop, but only the lease holder may write. + with PrecomputeLease(results_dir) as lease: + if lease: + # Dataset quality runs first because it is cheap and always finishes. + # The trends pass makes an LLM call per unprocessed run and gets + # SIGKILLed mid-pass on a large backlog, which the try/except below + # cannot catch and which would otherwise starve every pass after it. + for precompute in ( + precompute_dataset_quality.precompute, + precompute_trends.precompute, + ): + try: + precompute() + except Exception as e: + print(f"Error in {precompute.__module__}: {e}") time.sleep(interval) if __name__ == "__main__": diff --git a/viewer/trends.py b/viewer/trends.py index 1f1c7436..ab3a3aa2 100644 --- a/viewer/trends.py +++ b/viewer/trends.py @@ -3,6 +3,7 @@ import mesop as me import pandas as pd from main import State +from run_index import list_run_directories def get_results_dir(): # Try to read from environment variable @@ -103,12 +104,8 @@ def trends_component(): # Fallback to computing on the fly if cache is missing or failed if df is None: - directories = [ - d - for d in os.listdir(results_dir) - if os.path.isdir(os.path.join(results_dir, d)) - ] - + directories = list_run_directories(results_dir) + data = [] for d in directories: