diff --git a/docs/guides/portal-execution-profiles-handoff.md b/docs/guides/portal-execution-profiles-handoff.md index 1c28ea6..5252e2c 100644 --- a/docs/guides/portal-execution-profiles-handoff.md +++ b/docs/guides/portal-execution-profiles-handoff.md @@ -34,7 +34,9 @@ Recommended order: `RESULT_SERVER_GITLAB_TRIGGER_TOKEN`; do not store it in SQLite, logs, or the OSS repository. 6. Index received benchmark and estimation JSON metadata into SQLite while - keeping JSON/tgz artifacts as raw records. + keeping JSON/tgz artifacts as raw records. The first index should be an + auxiliary lookup table populated at ingest time; existing result and + estimate pages can remain file-backed until the indexed views are reviewed. 7. Add environment snapshot storage after deciding which host/runtime metadata should define an environment identity. diff --git a/result_server/routes/api.py b/result_server/routes/api.py index d74cfb9..207c688 100644 --- a/result_server/routes/api.py +++ b/result_server/routes/api.py @@ -4,6 +4,7 @@ import os import json import re +import sqlite3 import uuid import shutil import io @@ -15,6 +16,7 @@ from utils.auth import verify_ingest_key, verify_trusted_proxy_auth from utils.audit_logging import audit_event from utils.rate_limit import rate_limited +from utils.result_metadata_index import index_result_metadata api_bp = Blueprint("api", __name__) _TIMESTAMP_RE = re.compile(r"^\d{8}_\d{6}$") @@ -115,6 +117,41 @@ def save_json_file(data, prefix, out_dir, given_uuid=None): "id": unique_id, "timestamp": timestamp, "json_file": filename, + "payload": payload, + } + + +def _index_saved_json(record_type, saved): + """Index saved JSON metadata when a Portal SQLite DB is configured.""" + try: + indexed = index_result_metadata( + db_path=current_app.config.get("EXECUTION_PROFILE_DB_PATH"), + record_type=record_type, + payload=saved.get("payload", {}), + json_file=saved.get("json_file", ""), + fallback_uuid=saved.get("id", ""), + fallback_timestamp=saved.get("timestamp", ""), + ) + except (sqlite3.Error, OSError, ValueError) as exc: + current_app.logger.exception("result metadata index update failed") + audit_event( + "result_metadata_index_failed", + target=saved.get("json_file", ""), + result="failure", + level=logging.ERROR, + details={"record_type": record_type, "error": str(exc)}, + ) + return False + return indexed + + +def _saved_json_response(saved): + """Return the public API response fields for a saved JSON payload.""" + return { + "status": saved["status"], + "id": saved["id"], + "timestamp": saved["timestamp"], + "json_file": saved["json_file"], } @@ -280,6 +317,7 @@ def ingest_result(): prefix="result", out_dir=current_app.config["RECEIVED_DIR"], ) + _index_saved_json("result", saved) audit_event( "ingest_accepted", actor=runner_id, @@ -287,7 +325,7 @@ def ingest_result(): result="success", details={"ingest_type": "result", "id": saved["id"]}, ) - return saved, 200 + return _saved_json_response(saved), 200 @api_bp.route("/api/ingest/estimate", methods=["POST"]) @@ -306,6 +344,7 @@ def ingest_estimate(): out_dir=current_app.config["ESTIMATED_DIR"], given_uuid=raw_uuid, ) + _index_saved_json("estimate", saved) audit_event( "ingest_accepted", actor=runner_id, @@ -313,7 +352,7 @@ def ingest_estimate(): result="success", details={"ingest_type": "estimate", "id": saved["id"]}, ) - return saved, 200 + return _saved_json_response(saved), 200 @api_bp.route("/api/ingest/padata", methods=["POST"]) diff --git a/result_server/test_support.py b/result_server/test_support.py index 59f7c41..c165156 100644 --- a/result_server/test_support.py +++ b/result_server/test_support.py @@ -158,6 +158,7 @@ def build_api_route_app( received_padata_dir, received_estimation_artifacts_dir, estimated_dir, + execution_profile_db_path=None, ): """Build a Flask app with the API, results, and estimated blueprints for API tests.""" app = Flask(__name__) @@ -165,6 +166,8 @@ def build_api_route_app( app.config["RECEIVED_PADATA_DIR"] = received_padata_dir app.config["RECEIVED_ESTIMATION_ARTIFACTS_DIR"] = received_estimation_artifacts_dir app.config["ESTIMATED_DIR"] = estimated_dir + if execution_profile_db_path is not None: + app.config["EXECUTION_PROFILE_DB_PATH"] = execution_profile_db_path app.config["TESTING"] = True from routes.api import api_bp diff --git a/result_server/tests/test_api_routes.py b/result_server/tests/test_api_routes.py index c16c09e..73818d6 100644 --- a/result_server/tests/test_api_routes.py +++ b/result_server/tests/test_api_routes.py @@ -85,6 +85,60 @@ def test_post_valid_json(self, client, tmp_dirs): assert saved["code"] == "test" assert saved["_server_uuid"] == body["id"] assert saved["_server_timestamp"] == body["timestamp"] + assert "payload" not in body + + def test_post_valid_json_indexes_metadata(self, tmp_dirs, tmp_path): + """Accepted result payloads should update the Portal SQLite metadata index.""" + received, received_padata, received_estimation_artifacts, estimated = tmp_dirs + db_path = tmp_path / "cx_portal.sqlite3" + app = build_api_route_app( + received_dir=received, + received_padata_dir=received_padata, + received_estimation_artifacts_dir=received_estimation_artifacts, + estimated_dir=estimated, + execution_profile_db_path=str(db_path), + ) + app.config["INGEST_KEYS"] = {API_KEY: "test-runner"} + + with app.test_client() as client: + resp = client.post( + "/api/ingest/result", + data=json.dumps({ + "code": "qws", + "system": "RIKYU", + "Exp": "case0", + "FOM": 42.0, + "ci_trigger": "trigger", + "pipeline_id": 3152, + }), + headers={ + "X-API-Key": API_KEY, + "Content-Type": "application/json", + }, + ) + + assert resp.status_code == 200 + import sqlite3 + + with sqlite3.connect(db_path) as conn: + row = conn.execute( + """ + SELECT record_type, result_uuid, json_file, code, system, exp, + ci_trigger, pipeline_id + FROM result_metadata_index + """ + ).fetchone() + body = resp.get_json() + assert row == ( + "result", + body["id"], + body["json_file"], + "qws", + "RIKYU", + "case0", + "trigger", + "3152", + ) def test_valid_key_logs_runner_id(self, client, caplog): """Accepted API requests should include the resolved runner id in logs.""" @@ -209,6 +263,60 @@ def test_post_valid_json(self, client, tmp_dirs): assert saved["code"] == "est-test" assert saved["estimate_metadata"]["estimation_result_uuid"] == body["id"] assert "estimation_result_timestamp" in saved["estimate_metadata"] + assert "payload" not in body + + def test_post_valid_json_indexes_estimate_metadata(self, tmp_dirs, tmp_path): + """Accepted estimate payloads should update the Portal SQLite metadata index.""" + received, received_padata, received_estimation_artifacts, estimated = tmp_dirs + db_path = tmp_path / "cx_portal.sqlite3" + app = build_api_route_app( + received_dir=received, + received_padata_dir=received_padata, + received_estimation_artifacts_dir=received_estimation_artifacts, + estimated_dir=estimated, + execution_profile_db_path=str(db_path), + ) + app.config["INGEST_KEYS"] = {API_KEY: "test-runner"} + + with app.test_client() as client: + resp = client.post( + "/api/ingest/estimate", + data=json.dumps({ + "code": "qws", + "exp": "case0", + "performance_ratio": 1.5, + "current_system": {"system": "RIKYU"}, + "future_system": {"system": "FugakuNEXT"}, + "estimate_metadata": { + "source_result_uuid": "11111111-2222-3333-4444-555555555555", + "estimation_package": "weakscaling", + }, + }), + headers={ + "X-API-Key": API_KEY, + "Content-Type": "application/json", + }, + ) + + assert resp.status_code == 200 + import sqlite3 + + with sqlite3.connect(db_path) as conn: + row = conn.execute( + """ + SELECT record_type, result_uuid, json_file, code, system, exp + FROM result_metadata_index + """ + ).fetchone() + body = resp.get_json() + assert row == ( + "estimate", + body["id"], + body["json_file"], + "qws", + "RIKYU", + "case0", + ) def test_post_valid_json_with_uuid_header(self, client, tmp_dirs): """A valid X-UUID header should be used as the persisted estimate id.""" diff --git a/result_server/tests/test_result_metadata_index.py b/result_server/tests/test_result_metadata_index.py new file mode 100644 index 0000000..abe5306 --- /dev/null +++ b/result_server/tests/test_result_metadata_index.py @@ -0,0 +1,114 @@ +"""Tests for the SQLite result metadata index.""" + +from __future__ import annotations + +import json +import os +import sys + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +from utils.result_metadata_index import ( # noqa: E402 + extract_result_index_record, + index_result_metadata, + list_indexed_results, +) + + +def test_extract_result_index_record_from_benchmark_result(): + payload = { + "code": "qws", + "system": "RIKYU", + "Exp": "case0", + "FOM": 42.5, + "_server_uuid": "11111111-2222-3333-4444-555555555555", + "_server_timestamp": "20260806_010203", + "ci_trigger": "pipeline", + "pipeline_id": 3152, + "source_info": { + "source_type": "git", + "repo_url": "https://example.org/repo.git", + "branch": "develop", + "commit_hash": "abcdef123456", + }, + } + + record = extract_result_index_record( + record_type="result", + payload=payload, + json_file="result_20260806_010203_11111111-2222-3333-4444-555555555555.json", + ) + + assert record["result_uuid"] == "11111111-2222-3333-4444-555555555555" + assert record["server_timestamp"] == "20260806_010203" + assert record["code"] == "qws" + assert record["system"] == "RIKYU" + assert record["exp"] == "case0" + assert record["pipeline_id"] == "3152" + assert record["source_type"] == "git" + assert record["source_ref"] == "abcdef123456" + metadata = json.loads(record["metadata_json"]) + assert metadata["fom"] == 42.5 + assert metadata["source_info"]["branch"] == "develop" + + +def test_index_result_metadata_upserts_rows(tmp_path): + db_path = tmp_path / "cx_portal.sqlite3" + payload = { + "code": "qws", + "system": "RIKYU", + "_server_uuid": "11111111-2222-3333-4444-555555555555", + "_server_timestamp": "20260806_010203", + } + + indexed = index_result_metadata( + db_path=str(db_path), + record_type="result", + payload=payload, + json_file="result.json", + ) + payload["system"] = "FugakuNEXT" + index_result_metadata( + db_path=str(db_path), + record_type="result", + payload=payload, + json_file="result-renamed.json", + ) + + rows = list_indexed_results(str(db_path), record_type="result") + assert indexed is True + assert len(rows) == 1 + assert rows[0]["json_file"] == "result-renamed.json" + assert rows[0]["system"] == "FugakuNEXT" + + +def test_extract_result_index_record_from_estimate_result(): + payload = { + "code": "qws", + "exp": "case0", + "performance_ratio": 2.5, + "current_system": {"system": "RIKYU"}, + "future_system": {"system": "FugakuNEXT"}, + "estimate_metadata": { + "estimation_result_uuid": "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee", + "estimation_result_timestamp": "2026-08-06 01:02:03", + "source_result_uuid": "11111111-2222-3333-4444-555555555555", + "estimation_package": "weakscaling", + }, + "applicability": {"status": "applicable"}, + } + + record = extract_result_index_record( + record_type="estimate", + payload=payload, + json_file="estimate_20260806_010203_aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee.json", + ) + + assert record["result_uuid"] == "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee" + assert record["server_timestamp"] == "2026-08-06 01:02:03" + assert record["code"] == "qws" + assert record["system"] == "RIKYU" + assert record["exp"] == "case0" + metadata = json.loads(record["metadata_json"]) + assert metadata["source_result_uuid"] == "11111111-2222-3333-4444-555555555555" + assert metadata["future_system"] == "FugakuNEXT" diff --git a/result_server/utils/execution_profiles.py b/result_server/utils/execution_profiles.py index 976e41d..0556340 100644 --- a/result_server/utils/execution_profiles.py +++ b/result_server/utils/execution_profiles.py @@ -13,7 +13,7 @@ PROFILE_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:-]{0,127}$") -SCHEMA_VERSION = 2 +SCHEMA_VERSION = 3 @dataclass(frozen=True) @@ -162,6 +162,9 @@ def migrate(self) -> None: current = 1 if current < 2: self._apply_v2(conn) + current = 2 + if current < 3: + self._apply_v3(conn) def _apply_v1(self, conn: sqlite3.Connection) -> None: now = _utc_now_iso() @@ -236,6 +239,41 @@ def _apply_v2(self, conn: sqlite3.Connection) -> None: (2, now), ) + def _apply_v3(self, conn: sqlite3.Connection) -> None: + now = _utc_now_iso() + conn.executescript( + """ + CREATE TABLE result_metadata_index ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + record_type TEXT NOT NULL CHECK(record_type IN ('result', 'estimate')), + result_uuid TEXT NOT NULL, + server_timestamp TEXT NOT NULL DEFAULT '', + json_file TEXT NOT NULL, + code TEXT NOT NULL DEFAULT '', + system TEXT NOT NULL DEFAULT '', + exp TEXT NOT NULL DEFAULT '', + ci_trigger TEXT NOT NULL DEFAULT '', + pipeline_id TEXT NOT NULL DEFAULT '', + source_type TEXT NOT NULL DEFAULT '', + source_ref TEXT NOT NULL DEFAULT '', + metadata_json TEXT NOT NULL DEFAULT '{}', + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + UNIQUE(record_type, result_uuid), + UNIQUE(record_type, json_file) + ); + + CREATE INDEX idx_result_metadata_index_scope + ON result_metadata_index(record_type, code, system, exp); + CREATE INDEX idx_result_metadata_index_timestamp + ON result_metadata_index(record_type, server_timestamp); + """ + ) + conn.execute( + "INSERT INTO schema_migrations(version, applied_at) VALUES (?, ?)", + (3, now), + ) + def upsert_profile(self, profile: dict[str, Any], *, actor: str = "") -> None: self.migrate() now = _utc_now_iso() diff --git a/result_server/utils/result_metadata_index.py b/result_server/utils/result_metadata_index.py new file mode 100644 index 0000000..9f0d903 --- /dev/null +++ b/result_server/utils/result_metadata_index.py @@ -0,0 +1,217 @@ +"""SQLite metadata index for received benchmark and estimate JSON files.""" + +from __future__ import annotations + +import json +import os +import sqlite3 +from datetime import UTC, datetime +from typing import Any + +from utils.execution_profiles import ExecutionProfileStore + + +def _utc_now_iso() -> str: + return datetime.now(UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z") + + +def _as_text(value: Any) -> str: + if value is None: + return "" + if isinstance(value, (dict, list)): + return "" + return str(value).strip() + + +def _nested_text(data: dict[str, Any], *keys: str) -> str: + current: Any = data + for key in keys: + if not isinstance(current, dict): + return "" + current = current.get(key) + return _as_text(current) + + +def _source_ref(source_info: dict[str, Any]) -> str: + if not isinstance(source_info, dict): + return "" + for key in ("commit_hash", "md5sum", "branch", "file_path", "repo_url"): + value = _as_text(source_info.get(key)) + if value: + return value + return "" + + +def _metadata_for_result(payload: dict[str, Any]) -> dict[str, Any]: + source_info = payload.get("source_info") + source_info = source_info if isinstance(source_info, dict) else {} + return { + "fom": payload.get("FOM"), + "fom_unit": payload.get("FOM_unit"), + "fom_version": payload.get("FOM_version"), + "nodes": payload.get("nodes"), + "numproc_node": payload.get("numproc_node"), + "nthreads": payload.get("nthreads"), + "source_info": { + key: source_info.get(key) + for key in ("source_type", "repo_url", "branch", "commit_hash", "file_path", "md5sum") + if source_info.get(key) + }, + } + + +def _metadata_for_estimate(payload: dict[str, Any]) -> dict[str, Any]: + estimate_meta = payload.get("estimate_metadata") + estimate_meta = estimate_meta if isinstance(estimate_meta, dict) else {} + applicability = payload.get("applicability") + applicability = applicability if isinstance(applicability, dict) else {} + return { + "performance_ratio": payload.get("performance_ratio"), + "applicability_status": applicability.get("status"), + "estimation_package": estimate_meta.get("estimation_package"), + "requested_estimation_package": estimate_meta.get("requested_estimation_package"), + "source_result_uuid": estimate_meta.get("source_result_uuid"), + "source_result_timestamp": estimate_meta.get("source_result_timestamp"), + "current_system": payload.get("current_system", {}).get("system") + if isinstance(payload.get("current_system"), dict) + else None, + "future_system": payload.get("future_system", {}).get("system") + if isinstance(payload.get("future_system"), dict) + else None, + } + + +def extract_result_index_record( + *, + record_type: str, + payload: dict[str, Any], + json_file: str, + fallback_uuid: str = "", + fallback_timestamp: str = "", +) -> dict[str, Any]: + """Return a normalized result_metadata_index row from a stored JSON payload.""" + if record_type not in {"result", "estimate"}: + raise ValueError(f"unsupported record_type: {record_type}") + + if record_type == "estimate": + estimate_meta = payload.get("estimate_metadata") + estimate_meta = estimate_meta if isinstance(estimate_meta, dict) else {} + result_uuid = _as_text(estimate_meta.get("estimation_result_uuid")) or fallback_uuid + server_timestamp = ( + _as_text(estimate_meta.get("estimation_result_timestamp")) + or fallback_timestamp + ) + system = _nested_text(payload, "current_system", "system") + metadata = _metadata_for_estimate(payload) + else: + result_uuid = _as_text(payload.get("_server_uuid")) or fallback_uuid + server_timestamp = _as_text(payload.get("_server_timestamp")) or fallback_timestamp + system = _as_text(payload.get("system")) + metadata = _metadata_for_result(payload) + + source_info = payload.get("source_info") + source_info = source_info if isinstance(source_info, dict) else {} + return { + "record_type": record_type, + "result_uuid": result_uuid, + "server_timestamp": server_timestamp, + "json_file": os.path.basename(json_file), + "code": _as_text(payload.get("code")), + "system": system, + "exp": _as_text(payload.get("Exp") if record_type == "result" else payload.get("exp")), + "ci_trigger": _as_text(payload.get("ci_trigger")), + "pipeline_id": _as_text(payload.get("pipeline_id")), + "source_type": _as_text(source_info.get("source_type")), + "source_ref": _source_ref(source_info), + "metadata_json": json.dumps(metadata, ensure_ascii=False, sort_keys=True), + } + + +def index_result_metadata( + *, + db_path: str | None, + record_type: str, + payload: dict[str, Any], + json_file: str, + fallback_uuid: str = "", + fallback_timestamp: str = "", +) -> bool: + """Upsert a stored result or estimate JSON into the Portal SQLite index.""" + if not db_path: + return False + + record = extract_result_index_record( + record_type=record_type, + payload=payload, + json_file=json_file, + fallback_uuid=fallback_uuid, + fallback_timestamp=fallback_timestamp, + ) + if not record["result_uuid"]: + return False + + store = ExecutionProfileStore(db_path) + store.migrate() + now = _utc_now_iso() + with store.connect() as conn: + existing = conn.execute( + """ + SELECT created_at FROM result_metadata_index + WHERE record_type = ? AND result_uuid = ? + """, + (record["record_type"], record["result_uuid"]), + ).fetchone() + created_at = existing["created_at"] if existing else now + conn.execute( + """ + INSERT INTO result_metadata_index ( + record_type, result_uuid, server_timestamp, json_file, + code, system, exp, ci_trigger, pipeline_id, + source_type, source_ref, metadata_json, + created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(record_type, result_uuid) DO UPDATE SET + server_timestamp=excluded.server_timestamp, + json_file=excluded.json_file, + code=excluded.code, + system=excluded.system, + exp=excluded.exp, + ci_trigger=excluded.ci_trigger, + pipeline_id=excluded.pipeline_id, + source_type=excluded.source_type, + source_ref=excluded.source_ref, + metadata_json=excluded.metadata_json, + updated_at=excluded.updated_at + """, + ( + record["record_type"], + record["result_uuid"], + record["server_timestamp"], + record["json_file"], + record["code"], + record["system"], + record["exp"], + record["ci_trigger"], + record["pipeline_id"], + record["source_type"], + record["source_ref"], + record["metadata_json"], + created_at, + now, + ), + ) + return True + + +def list_indexed_results(db_path: str, *, record_type: str | None = None) -> list[dict[str, Any]]: + """Return indexed metadata rows, newest first.""" + store = ExecutionProfileStore(db_path) + store.migrate() + query = "SELECT * FROM result_metadata_index" + params: tuple[str, ...] = () + if record_type: + query += " WHERE record_type = ?" + params = (record_type,) + query += " ORDER BY server_timestamp DESC, id DESC" + with store.connect() as conn: + return [dict(row) for row in conn.execute(query, params).fetchall()]