diff --git a/fournos-ui/.dockerignore b/fournos-ui/.dockerignore new file mode 100644 index 0000000..e632277 --- /dev/null +++ b/fournos-ui/.dockerignore @@ -0,0 +1,19 @@ +__pycache__ +*.pyc +*.pyo +.git +.gitignore +.env +.env.* +*.md +.venv +venv +.mypy_cache +.pytest_cache +.ruff_cache +.DS_Store + +app/mock_data.py + +kustomize/ +resolvers/ diff --git a/fournos-ui/.gitignore b/fournos-ui/.gitignore new file mode 100644 index 0000000..bb12cb0 --- /dev/null +++ b/fournos-ui/.gitignore @@ -0,0 +1,36 @@ +# Secrets - never commit actual credentials +kustomize/overlays/*/postgresql-secret.env +*secret*.env + +# Python +__pycache__/ +*.pyc +*.pyo +.venv/ +venv/ + +# Database files +*.db +*.sqlite +*.sqlite3 + +# IDE +.idea/ +.vscode/ + +# Environment +.env +.env.* +!.env.example + +# OS +.DS_Store + +# Dev-only +app/mock_data.py + +# Local overlay overrides (users create their own from *.example) +kustomize/overlays/*/params.env +kustomize/overlays/*/projects.yaml +kustomize/overlays/*/kustomization.yaml +kustomize/overlays/*/rolebinding-*.yaml diff --git a/fournos-ui/Dockerfile b/fournos-ui/Dockerfile new file mode 100644 index 0000000..6f10014 --- /dev/null +++ b/fournos-ui/Dockerfile @@ -0,0 +1,23 @@ +FROM python:3.12-slim AS base + +RUN groupadd --gid 1001 app && \ + useradd --uid 1001 --gid app --create-home app + +WORKDIR /opt/fournos-dashboard + +COPY requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt + +COPY app/ app/ + +RUN chown -R app:app /opt/fournos-dashboard + +USER app + +EXPOSE 8000 + +ENTRYPOINT ["uvicorn", "app.main:app", \ + "--host", "0.0.0.0", \ + "--port", "8000", \ + "--workers", "1", \ + "--log-level", "info"] diff --git a/fournos-ui/README.md b/fournos-ui/README.md new file mode 100644 index 0000000..1df9566 --- /dev/null +++ b/fournos-ui/README.md @@ -0,0 +1,136 @@ +# Fournos Dashboard + +A web dashboard for managing [Fournos](https://github.com/openshift-psap/fournos-operator) performance testing jobs on Kubernetes. Submit jobs, monitor live runs, schedule recurring tests, and review historical results -- all from one place. + +## What It Does + +- **Live job monitoring** -- Watch running FournosJobs with real-time log streaming (SSE) and pipeline progress tracking. +- **Job submission** -- Submit new FournosJobs with project, preset, cluster, and config override selection. Optionally pick an open Forge PR to test with -- the dashboard fetches open PRs from GitHub and fills in the commit SHA automatically. +- **GitHub PR integration** -- Lists open pull requests from the [Forge repo](https://github.com/openshift-psap/forge) directly in the submit form. Since Forge is a public repository, no GitHub token is needed. +- **Scheduling** -- Create Kubernetes CronJobs for recurring test runs, with optional version-resolver scripts that dynamically determine parameters at runtime. +- **History** -- Browse completed jobs stored in PostgreSQL with status, duration, and direct links to MLflow artifacts. +- **Schedule tracking** -- See which schedule triggered each job (manual vs. scheduled) and view all runs for a given schedule. + +## Architecture + +``` +┌─────────────┐ ┌──────────────────┐ ┌────────────┐ +│ Browser │────▶│ FastAPI + HTMX │────▶│ Kubernetes │ +│ │◀────│ (Dashboard) │◀────│ API │ +└─────────────┘ └────────┬─────────┘ └────────────┘ + │ + ┌────────▼─────────┐ + │ PostgreSQL │ + │ (job history) │ + └──────────────────┘ +``` + +- **FastAPI** backend with **Jinja2** templates and **HTMX** for dynamic updates. +- **Kubernetes Python client** for watching FournosJob CRs, streaming pod logs, and managing CronJobs. +- **PostgreSQL** (via SQLAlchemy async + asyncpg) for persisting job metadata and schedule tracking. +- A background **watcher thread** monitors FournosJob events and archives them to PostgreSQL automatically. + +## Prerequisites + +- A Kubernetes / OpenShift cluster with the [Fournos Operator](https://github.com/openshift-psap/fournos-operator) installed. +- A container registry to push the dashboard image. +- `kubectl` or `oc` CLI configured with cluster access. + +## Getting Started + +### 1. Clone and configure the overlay + +```bash +cd kustomize/overlays/ocp/ + +# Copy example files +cp kustomization.yaml.example kustomization.yaml +cp projects.yaml.example projects.yaml +cp params.env.example params.env +cp ../../base/postgresql-secret.env.example postgresql-secret.env +``` + +Edit each file with your values: +- **`kustomization.yaml`** -- Set your dashboard image, PostgreSQL image, target namespace, and storage class. +- **`projects.yaml`** -- Define your Forge projects, clusters, and presets. +- **`postgresql-secret.env`** -- Set your database credentials. +- **`params.env`** -- Set your storage class and size. + +### 2. Build and push the dashboard image + +### 3. Deploy to the cluster + +```bash +cd kustomize/overlays/ocp/ + +# Apply the main stack +oc kustomize . | oc apply -f - + +# Apply the cross-namespace RoleBinding (grants dashboard access to the jobs namespace) +oc apply -f rolebinding-psap-automation.yaml +``` + +This creates: +- A `fournos-dashboard` namespace +- PostgreSQL StatefulSet with persistent storage +- Dashboard Deployment, Service, ServiceAccount +- ClusterRole for FournosJob/CronJob/Pod access +- RoleBinding in the target namespace (e.g. `psap-automation`) +- Projects ConfigMap + +### 4. Access the dashboard + +```bash +oc port-forward -n fournos-dashboard svc/fournos-dashboard 8000:8000 +``` + +Open http://localhost:8000 + + +## Configuration + +All configuration is via environment variables (set in the deployment manifest): + +| Variable | Description | Default | +|---|---|---| +| `DATABASE_URL` | PostgreSQL connection string (required) | *none -- must be set* | +| `FOURNOS_NAMESPACE` | Namespace where FournosJobs run | *set via overlay* | +| `PROJECTS_CONFIG_PATH` | Path to projects YAML | `/etc/fournos-dashboard/projects.yaml` | +| `K8S_REQUEST_TIMEOUT` | Timeout for K8s API calls (seconds) | `30` | +| `LOG_LEVEL` | Logging level | `INFO` | +| `KUBECONFIG` | Path to kubeconfig (local dev only) | in-cluster config | +| `FORGE_GITHUB_REPO` | GitHub `owner/repo` for PR listing | `openshift-psap/forge` | + +## Security Considerations + +This dashboard is designed as an **internal tool** and does **not** include built-in authentication or authorization. As described above, the tool is accessible when port-forwarding from the cluster where it's running. Future development may include auth. + +## Local Development + +```bash +pip install -r requirements.txt + +# Set DATABASE_URL and KUBECONFIG, then: +uvicorn app.main:app --reload --port 8000 +``` + +## Project Structure + +``` +fournos-ui/ +├── app/ +│ ├── main.py # FastAPI routes and Jinja2 rendering +│ ├── config.py # Environment-based settings +│ ├── db.py # SQLAlchemy models and queries +│ ├── k8s_client.py # Kubernetes API wrapper (with timeouts) +│ ├── watcher.py # Background FournosJob event watcher +│ ├── forge_discovery.py # Project discovery from ConfigMap +│ ├── models.py # Pydantic/dataclass models +│ ├── static/ # CSS, HTMX, htmx-sse.js +│ └── templates/ # Jinja2 HTML templates +├── kustomize/ +│ ├── base/ # Generic K8s manifests +│ └── overlays/ocp/ # Environment-specific overrides +├── Dockerfile +└── requirements.txt +``` diff --git a/fournos-ui/app/__init__.py b/fournos-ui/app/__init__.py new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/fournos-ui/app/__init__.py @@ -0,0 +1 @@ + diff --git a/fournos-ui/app/config.py b/fournos-ui/app/config.py new file mode 100644 index 0000000..aeef2e2 --- /dev/null +++ b/fournos-ui/app/config.py @@ -0,0 +1,62 @@ +"""Application configuration loaded from environment variables.""" + +from __future__ import annotations + +import os +from dataclasses import dataclass, field + + +@dataclass(frozen=True) +class Settings: + database_url: str = field( + default_factory=lambda: os.environ["DATABASE_URL"] + ) + + fournos_namespace: str = field( + default_factory=lambda: os.environ.get("FOURNOS_NAMESPACE", "fournos-jobs") + ) + + kubeconfig_path: str | None = field( + default_factory=lambda: os.environ.get("KUBECONFIG") + ) + + forge_repo_path: str | None = field( + default_factory=lambda: os.environ.get("FORGE_REPO_PATH") + ) + + projects_config_path: str = field( + default_factory=lambda: os.environ.get("PROJECTS_CONFIG_PATH", "/etc/fournos-dashboard/projects.yaml") + ) + + fournos_api_group: str = "fournos.dev" + fournos_api_version: str = "v1" + fournos_job_plural: str = "fournosjobs" + + tekton_api_group: str = "tekton.dev" + tekton_api_version: str = "v1" + tekton_pipelinerun_plural: str = "pipelineruns" + + log_level: str = field( + default_factory=lambda: os.environ.get("LOG_LEVEL", "INFO") + ) + + forge_github_repo: str = field( + default_factory=lambda: os.environ.get("FORGE_GITHUB_REPO", "openshift-psap/forge") + ) + + k8s_request_timeout_seconds: int = field( + default_factory=lambda: int(os.environ.get("K8S_REQUEST_TIMEOUT", "30")) + ) + + jobs_poll_interval_seconds: int = 5 + + default_pipelines: tuple[str, ...] = ( + "forge-full", + "forge-prepare-test", + "forge-test-only", + "forge-prepare-only", + "forge-replot", + ) + + +settings = Settings() diff --git a/fournos-ui/app/db.py b/fournos-ui/app/db.py new file mode 100644 index 0000000..b009fe4 --- /dev/null +++ b/fournos-ui/app/db.py @@ -0,0 +1,193 @@ +"""PostgreSQL persistence layer using SQLAlchemy async.""" + +from __future__ import annotations + +import logging +from datetime import datetime, timezone +from typing import Any, Sequence +from uuid import uuid4 + +from sqlalchemy import ( + Column, + DateTime, + Float, + ForeignKey, + String, + Text, + func, + select, +) +from sqlalchemy.dialects.postgresql import ARRAY, JSONB, insert as pg_insert +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine +from sqlalchemy.orm import DeclarativeBase, relationship + +from app.config import settings + +logger = logging.getLogger(__name__) + + +class Base(DeclarativeBase): + pass + + +# --------------------------------------------------------------------------- +# ORM Models +# --------------------------------------------------------------------------- + +class Job(Base): + __tablename__ = "jobs" + + id = Column(String, primary_key=True, default=lambda: str(uuid4())) + name = Column(String, unique=True, nullable=False, index=True) + project = Column(String, nullable=False, index=True) + preset = Column(String, default="") + cluster = Column(String, nullable=False, index=True) + pipeline = Column(String, default="") + owner = Column(String, default="", index=True) + status = Column(String, default="Pending", index=True) + message = Column(Text, default="") + created_at = Column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)) + completed_at = Column(DateTime(timezone=True), nullable=True) + duration_seconds = Column(Float, nullable=True) + mlflow_url = Column(String, default="") + ci_artifacts_url = Column(String, default="") + config_overrides = Column(JSONB, default=dict) + tags = Column(ARRAY(String), default=list) + fjob_spec = Column(JSONB, default=dict) + fjob_status = Column(JSONB, default=dict) + error_message = Column(Text, default="") + triggered_by_schedule = Column(String, nullable=True, index=True) + trigger_type = Column(String, default="manual") + + events = relationship("JobEvent", back_populates="job", cascade="all, delete-orphan") + + +class JobEvent(Base): + __tablename__ = "job_events" + + id = Column(String, primary_key=True, default=lambda: str(uuid4())) + job_id = Column(String, ForeignKey("jobs.id", ondelete="CASCADE"), nullable=False, index=True) + phase = Column(String, nullable=False) + message = Column(Text, default="") + timestamp = Column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)) + + job = relationship("Job", back_populates="events") + + +# --------------------------------------------------------------------------- +# Engine & session factory +# --------------------------------------------------------------------------- + +engine = create_async_engine(settings.database_url, echo=False, pool_size=5, max_overflow=10) +async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) + + +async def init_db() -> None: + """Create all tables (development convenience -- use Alembic in production).""" + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + logger.info("Database tables ensured") + + +# --------------------------------------------------------------------------- +# Query helpers +# --------------------------------------------------------------------------- + +async def upsert_job(session: AsyncSession, **kwargs: Any) -> Job: + """Insert or update a job record keyed by name (atomic).""" + if "id" not in kwargs: + kwargs["id"] = str(uuid4()) + + update_cols = {k: v for k, v in kwargs.items() if k not in ("id", "name") and v is not None} + + stmt = ( + pg_insert(Job) + .values(**kwargs) + .on_conflict_do_update(index_elements=["name"], set_=update_cols) + .returning(Job) + ) + result = await session.execute(stmt) + job = result.scalar_one() + return job + + +async def add_job_event( + session: AsyncSession, job_id: str, phase: str, message: str = "" +) -> JobEvent: + """Record a status transition.""" + event = JobEvent(job_id=job_id, phase=phase, message=message) + session.add(event) + await session.flush() + return event + + +async def get_job_by_name(session: AsyncSession, name: str) -> Job | None: + result = await session.execute(select(Job).where(Job.name == name)) + return result.scalar_one_or_none() + + +async def list_jobs( + session: AsyncSession, + *, + project: str | None = None, + cluster: str | None = None, + status: str | None = None, + owner: str | None = None, + limit: int = 50, + offset: int = 0, +) -> tuple[Sequence[Job], int]: + """List archived jobs with optional filters. Returns (jobs, total_count).""" + stmt = select(Job) + count_stmt = select(func.count(Job.id)) + + if project: + stmt = stmt.where(Job.project == project) + count_stmt = count_stmt.where(Job.project == project) + if cluster: + stmt = stmt.where(Job.cluster == cluster) + count_stmt = count_stmt.where(Job.cluster == cluster) + if status: + stmt = stmt.where(Job.status == status) + count_stmt = count_stmt.where(Job.status == status) + if owner: + stmt = stmt.where(Job.owner == owner) + count_stmt = count_stmt.where(Job.owner == owner) + + stmt = stmt.order_by(Job.created_at.desc()).limit(limit).offset(offset) + + result = await session.execute(stmt) + jobs = result.scalars().all() + + count_result = await session.execute(count_stmt) + total = count_result.scalar() or 0 + + return jobs, total + + +async def list_jobs_by_schedule( + session: AsyncSession, schedule_name: str, +) -> Sequence[Job]: + """List all jobs triggered by a specific schedule.""" + result = await session.execute( + select(Job) + .where(Job.triggered_by_schedule == schedule_name) + .order_by(Job.created_at.desc()) + ) + return result.scalars().all() + + +async def delete_job_by_name(session: AsyncSession, name: str) -> bool: + """Delete a job and all related logs/events by name. Returns True if deleted.""" + job = await get_job_by_name(session, name) + if job is None: + return False + await session.delete(job) + await session.flush() + return True + + +async def get_job_events(session: AsyncSession, job_id: str) -> Sequence[JobEvent]: + result = await session.execute( + select(JobEvent).where(JobEvent.job_id == job_id).order_by(JobEvent.timestamp) + ) + return result.scalars().all() diff --git a/fournos-ui/app/forge_discovery.py b/fournos-ui/app/forge_discovery.py new file mode 100644 index 0000000..99715ff --- /dev/null +++ b/fournos-ui/app/forge_discovery.py @@ -0,0 +1,161 @@ +"""Discover Forge projects, presets, and config schemas from the repo.""" + +from __future__ import annotations + +import logging +from pathlib import Path + +import yaml + +from app.config import settings +from app.models import ProjectInfo + +logger = logging.getLogger(__name__) + +_cache: dict[str, ProjectInfo] | None = None + + +def _forge_projects_dir() -> Path | None: + """Resolve the forge projects/ directory.""" + if settings.forge_repo_path: + p = Path(settings.forge_repo_path) / "projects" + if p.is_dir(): + return p + return None + + +def discover_projects(force_refresh: bool = False) -> list[ProjectInfo]: + """Discover projects from Forge repo or ConfigMap-backed YAML file.""" + global _cache + if _cache is not None and not force_refresh: + return list(_cache.values()) + + result: dict[str, ProjectInfo] = {} + + projects_dir = _forge_projects_dir() + if projects_dir is not None: + result = _discover_from_repo(projects_dir) + + if not result: + result = _discover_from_configmap() + + _cache = result + logger.info("Discovered %d Forge projects", len(result)) + return list(result.values()) + + +def _discover_from_repo(projects_dir: Path) -> dict[str, ProjectInfo]: + """Scan the Forge repo for available projects.""" + result: dict[str, ProjectInfo] = {} + skip = {"core", "__pycache__"} + + for proj_dir in sorted(projects_dir.iterdir()): + if not proj_dir.is_dir() or proj_dir.name.startswith(".") or proj_dir.name in skip: + continue + + orchestration = proj_dir / "orchestration" + if not orchestration.is_dir(): + continue + + presets = _load_presets(orchestration) + config_keys = _load_config_keys(orchestration) + has_cli = (orchestration / "cli.py").exists() + + result[proj_dir.name] = ProjectInfo( + name=proj_dir.name, + presets=presets, + config_keys=config_keys, + has_cli=has_cli, + ) + + return result + + +def _discover_from_configmap() -> dict[str, ProjectInfo]: + """Load project definitions from a YAML config file (mounted from a ConfigMap).""" + config_path = Path(settings.projects_config_path) + if not config_path.exists(): + logger.warning("No projects config at %s", config_path) + return {} + + try: + with open(config_path) as f: + data = yaml.safe_load(f) + except Exception as exc: + logger.error("Failed to parse projects config: %s", exc) + return {} + + if not isinstance(data, dict) or "projects" not in data: + logger.warning("Projects config missing 'projects' key") + return {} + + result: dict[str, ProjectInfo] = {} + for idx, proj in enumerate(data["projects"]): + if not isinstance(proj, dict): + logger.warning("Skipping malformed project entry at index %d: expected mapping, got %s", idx, type(proj).__name__) + continue + name = proj.get("name", "") + if not name: + continue + result[name] = ProjectInfo( + name=name, + cluster=proj.get("cluster", ""), + presets=proj.get("presets", []), + config_keys=proj.get("config_keys", []), + has_cli=proj.get("has_cli", False), + ) + + logger.info("Loaded %d projects from ConfigMap", len(result)) + return result + + +def get_project(name: str) -> ProjectInfo | None: + """Get info for a specific project.""" + projects = discover_projects() + return next((p for p in projects if p.name == name), None) + + +def get_project_presets(name: str) -> list[str]: + """Get available presets for a project.""" + proj = get_project(name) + return proj.presets if proj else [] + + +def _load_presets(orchestration_dir: Path) -> list[str]: + """Load preset names from presets.d/.""" + presets_dir = orchestration_dir / "presets.d" + if not presets_dir.is_dir(): + return [] + + preset_names: list[str] = [] + for yaml_file in sorted(presets_dir.glob("*.yaml")): + try: + with open(yaml_file) as f: + data = yaml.safe_load(f) + if isinstance(data, dict): + preset_names.extend(data.keys()) + except Exception as exc: + logger.debug("Failed to parse %s: %s", yaml_file, exc) + return preset_names + + +def _load_config_keys(orchestration_dir: Path) -> list[str]: + """Extract top-level config keys from config.yaml and config.d/.""" + keys: list[str] = [] + + config_file = orchestration_dir / "config.yaml" + if config_file.exists(): + try: + with open(config_file) as f: + data = yaml.safe_load(f) + if isinstance(data, dict): + keys.extend(data.keys()) + except Exception: + pass + + config_d = orchestration_dir / "config.d" + if config_d.is_dir(): + for yaml_file in sorted(config_d.glob("*.yaml")): + keys.append(yaml_file.stem) + + return keys diff --git a/fournos-ui/app/k8s_client.py b/fournos-ui/app/k8s_client.py new file mode 100644 index 0000000..abfce55 --- /dev/null +++ b/fournos-ui/app/k8s_client.py @@ -0,0 +1,849 @@ +"""Kubernetes client wrapper for FournosJob, PipelineRun, Pod, and CronJob operations.""" + +from __future__ import annotations + +import json +import logging +import threading +from datetime import datetime, timezone +from typing import Any, Generator + +import yaml +from kubernetes import client, config, watch +from kubernetes.client.rest import ApiException + +from app.config import settings + +logger = logging.getLogger(__name__) + +_api_client: client.ApiClient | None = None +_custom_api: client.CustomObjectsApi | None = None +_core_api: client.CoreV1Api | None = None +_batch_api: client.BatchV1Api | None = None +_lock = threading.Lock() + + +def _ensure_loaded() -> None: + """Load kubeconfig or in-cluster config once.""" + global _api_client, _custom_api, _core_api, _batch_api + if _custom_api is not None: + return + with _lock: + if _custom_api is not None: + return + try: + if settings.kubeconfig_path: + config.load_kube_config(config_file=settings.kubeconfig_path) + else: + try: + config.load_incluster_config() + except config.ConfigException: + config.load_kube_config() + except Exception: + logger.warning("K8s config not available -- running in offline mode") + return + + configuration = client.Configuration.get_default_copy() + timeout = settings.k8s_request_timeout_seconds + configuration.connect_timeout = timeout + configuration.read_timeout = timeout + _api_client = client.ApiClient(configuration=configuration) + _custom_api = client.CustomObjectsApi(_api_client) + _core_api = client.CoreV1Api(_api_client) + _batch_api = client.BatchV1Api(_api_client) + logger.info("Kubernetes client initialised (timeout=%ds)", timeout) + + +def is_connected() -> bool: + """Return True if a K8s client has been successfully loaded.""" + _ensure_loaded() + return _custom_api is not None + + +# --------------------------------------------------------------------------- +# FournosJob operations +# --------------------------------------------------------------------------- + +def list_fournos_jobs(namespace: str | None = None) -> list[dict]: + """List all FournosJob CRs in the given namespace.""" + _ensure_loaded() + if _custom_api is None: + return [] + ns = namespace or settings.fournos_namespace + try: + result = _custom_api.list_namespaced_custom_object( + group=settings.fournos_api_group, + version=settings.fournos_api_version, + namespace=ns, + plural=settings.fournos_job_plural, + ) + return result.get("items", []) + except ApiException as exc: + logger.error("Failed to list FournosJobs: %s", exc.reason) + return [] + + +def get_fournos_job(name: str, namespace: str | None = None) -> dict | None: + """Get a specific FournosJob by name.""" + _ensure_loaded() + if _custom_api is None: + return None + ns = namespace or settings.fournos_namespace + try: + return _custom_api.get_namespaced_custom_object( + group=settings.fournos_api_group, + version=settings.fournos_api_version, + namespace=ns, + plural=settings.fournos_job_plural, + name=name, + ) + except ApiException as exc: + if exc.status == 404: + return None + logger.error("Failed to get FournosJob %s: %s", name, exc.reason) + return None + + +def create_fournos_job(body: dict, namespace: str | None = None) -> dict: + """Create a new FournosJob CR.""" + _ensure_loaded() + if _custom_api is None: + raise RuntimeError("Kubernetes client not available") + ns = namespace or settings.fournos_namespace + return _custom_api.create_namespaced_custom_object( + group=settings.fournos_api_group, + version=settings.fournos_api_version, + namespace=ns, + plural=settings.fournos_job_plural, + body=body, + ) + + +def patch_fournos_job( + name: str, patch: dict, namespace: str | None = None +) -> dict: + """Patch a FournosJob (e.g. set spec.shutdown).""" + _ensure_loaded() + if _custom_api is None: + raise RuntimeError("Kubernetes client not available") + ns = namespace or settings.fournos_namespace + return _custom_api.patch_namespaced_custom_object( + group=settings.fournos_api_group, + version=settings.fournos_api_version, + namespace=ns, + plural=settings.fournos_job_plural, + name=name, + body=patch, + ) + + +def shutdown_fournos_job( + name: str, value: str = "Stop", namespace: str | None = None +) -> dict: + """Set spec.shutdown on a FournosJob to cancel it.""" + return patch_fournos_job(name, {"spec": {"shutdown": value}}, namespace) + + +def watch_fournos_jobs( + namespace: str | None = None, + resource_version: str = "", + timeout: int = 0, +) -> Generator[dict, None, None]: + """Yield watch events for FournosJobs. Blocks until timeout or stream ends.""" + _ensure_loaded() + if _custom_api is None: + return + ns = namespace or settings.fournos_namespace + w = watch.Watch() + kwargs: dict[str, Any] = { + "group": settings.fournos_api_group, + "version": settings.fournos_api_version, + "namespace": ns, + "plural": settings.fournos_job_plural, + } + if resource_version: + kwargs["resource_version"] = resource_version + if timeout: + kwargs["timeout_seconds"] = timeout + try: + for event in w.stream(_custom_api.list_namespaced_custom_object, **kwargs): + yield event + except ApiException as exc: + logger.warning("Watch stream ended: %s", exc.reason) + + +# --------------------------------------------------------------------------- +# Tekton PipelineRun operations +# --------------------------------------------------------------------------- + +def get_pipelinerun(name: str, namespace: str | None = None) -> dict | None: + """Get a Tekton PipelineRun by name.""" + _ensure_loaded() + if _custom_api is None: + return None + ns = namespace or settings.fournos_namespace + try: + return _custom_api.get_namespaced_custom_object( + group=settings.tekton_api_group, + version=settings.tekton_api_version, + namespace=ns, + plural=settings.tekton_pipelinerun_plural, + name=name, + ) + except ApiException as exc: + if exc.status == 404: + return None + logger.error("Failed to get PipelineRun %s: %s", name, exc.reason) + return None + + +def list_pipelineruns_for_job( + job_name: str, namespace: str | None = None +) -> list[dict]: + """List PipelineRuns associated with a FournosJob (by label).""" + _ensure_loaded() + if _custom_api is None: + return [] + ns = namespace or settings.fournos_namespace + try: + result = _custom_api.list_namespaced_custom_object( + group=settings.tekton_api_group, + version=settings.tekton_api_version, + namespace=ns, + plural=settings.tekton_pipelinerun_plural, + label_selector=f"fournos.dev/job-name={job_name}", + ) + return result.get("items", []) + except ApiException as exc: + logger.error("Failed to list PipelineRuns for %s: %s", job_name, exc.reason) + return [] + + +def get_taskrun(name: str, namespace: str | None = None) -> dict | None: + """Get a Tekton TaskRun by name.""" + _ensure_loaded() + if _custom_api is None: + return None + ns = namespace or settings.fournos_namespace + try: + return _custom_api.get_namespaced_custom_object( + group=settings.tekton_api_group, + version=settings.tekton_api_version, + namespace=ns, + plural="taskruns", + name=name, + ) + except ApiException as exc: + if exc.status == 404: + return None + logger.error("Failed to get TaskRun %s: %s", name, exc.reason) + return None + + +def _phase_from_conditions(conditions: list[dict]) -> str: + """Determine task phase from Tekton conditions.""" + if not conditions: + return "Pending" + cond = conditions[0] + reason = cond.get("reason", "") + cond_status = cond.get("status", "") + if reason == "Succeeded" and cond_status == "True": + return "Succeeded" + if reason == "Failed" or cond_status == "False": + return "Failed" + if reason in ("Running", "Started"): + return "Running" + if reason == "TaskRunCancelled": + return "Cancelled" + if reason == "SkippingNoMatch": + return "Skipped" + return "Pending" + + +def get_current_step_for_job( + job_name: str, namespace: str | None = None +) -> dict | None: + """Return the currently running pipeline step for a job, or None. + + Result dict has keys: name, displayName, startTime. + """ + prs = list_pipelineruns_for_job(job_name, namespace) + if not prs: + return None + + child_refs = prs[0].get("status", {}).get("childReferences", []) + for ref in child_refs: + task_run_name = ref.get("name", "") + tr = get_taskrun(task_run_name) + if not tr: + continue + tr_status = tr.get("status", {}) + phase = _phase_from_conditions(tr_status.get("conditions", [])) + if phase == "Running": + task_name = ref.get("pipelineTaskName", task_run_name) + return { + "name": task_name, + "displayName": task_name.replace("-", " ").title(), + "startTime": tr_status.get("startTime"), + } + return None + + +def extract_pipeline_stages(pipelinerun: dict) -> list[dict]: + """Extract stage information from a PipelineRun status for timeline display.""" + status = pipelinerun.get("status", {}) + child_refs = status.get("childReferences", []) + pipeline_spec = status.get("pipelineSpec", {}) + + finally_task_names = set() + for task in pipeline_spec.get("finally", []): + finally_task_names.add(task.get("name", "")) + + stages = [] + for ref in child_refs: + task_name = ref.get("pipelineTaskName", ref.get("name", "unknown")) + task_run_name = ref.get("name", "") + + start_time = None + completion_time = None + task_phase = "Pending" + + tr = get_taskrun(task_run_name) + if tr: + tr_status = tr.get("status", {}) + start_time = tr_status.get("startTime") + completion_time = tr_status.get("completionTime") + task_phase = _phase_from_conditions(tr_status.get("conditions", [])) + + display_name = task_name.replace("-", " ").title() + + stages.append({ + "name": task_name, + "displayName": display_name, + "status": task_phase, + "startTime": start_time, + "completionTime": completion_time, + "finally": task_name in finally_task_names, + }) + + stages.sort(key=lambda s: (s["finally"], s.get("startTime") or "9999")) + return stages + + +# --------------------------------------------------------------------------- +# Pod operations +# --------------------------------------------------------------------------- + +def list_pods_for_job(job_name: str, namespace: str | None = None) -> list[dict]: + """List pods associated with a FournosJob.""" + _ensure_loaded() + if _core_api is None: + return [] + ns = namespace or settings.fournos_namespace + try: + result = _core_api.list_namespaced_pod( + namespace=ns, + label_selector=f"fournos.dev/job-name={job_name}", + ) + pods = [] + for pod in result.items: + created = pod.metadata.creation_timestamp + age_minutes = 0 + if created: + delta = datetime.now(timezone.utc) - created.replace(tzinfo=timezone.utc) + age_minutes = int(delta.total_seconds() / 60) + + container_ready = False + restarts = 0 + if pod.status.container_statuses: + for cs in pod.status.container_statuses: + if cs.ready: + container_ready = True + restarts += cs.restart_count + + if pod.metadata.name.startswith("affinity-assistant"): + continue + + pods.append({ + "name": pod.metadata.name, + "phase": pod.status.phase or "Unknown", + "container": ( + pod.spec.containers[0].name if pod.spec.containers else "unknown" + ), + "ready": container_ready, + "restarts": restarts, + "age_minutes": age_minutes, + "_created": created, + }) + pods.sort(key=lambda p: p["_created"] or datetime.min.replace(tzinfo=timezone.utc)) + return pods + except ApiException as exc: + logger.error("Failed to list pods for %s: %s", job_name, exc.reason) + return [] + + +def read_pod_log( + pod_name: str, + namespace: str | None = None, + container: str | None = None, + follow: bool = False, + tail_lines: int | None = None, +) -> Generator[str, None, None]: + """Stream or read pod logs line by line.""" + _ensure_loaded() + if _core_api is None: + yield "Kubernetes client not available" + return + ns = namespace or settings.fournos_namespace + kwargs: dict[str, Any] = {"name": pod_name, "namespace": ns, "follow": follow} + if container: + kwargs["container"] = container + if tail_lines: + kwargs["tail_lines"] = tail_lines + try: + if follow: + for line in _core_api.read_namespaced_pod_log(**kwargs, _preload_content=False).stream(): + decoded = line.decode("utf-8", errors="replace").rstrip("\n") + yield decoded + else: + log_text = _core_api.read_namespaced_pod_log(**kwargs) + for line in log_text.splitlines(): + yield line + except ApiException as exc: + yield f"Error reading logs: {exc.reason}" + + +def read_pod_log_full( + pod_name: str, + namespace: str | None = None, + container: str | None = None, +) -> str: + """Read entire pod log as a single string (for archival).""" + _ensure_loaded() + if _core_api is None: + return "" + ns = namespace or settings.fournos_namespace + kwargs: dict[str, Any] = {"name": pod_name, "namespace": ns} + if container: + kwargs["container"] = container + try: + return _core_api.read_namespaced_pod_log(**kwargs) + except ApiException as exc: + logger.error("Failed to read full log for %s: %s", pod_name, exc.reason) + return f"Error: {exc.reason}" + + +# --------------------------------------------------------------------------- +# CronJob operations (for scheduling) +# --------------------------------------------------------------------------- + +# --------------------------------------------------------------------------- + +SCHEDULE_LABEL = "fournos-launcher/managed-by" +SCHEDULE_LABEL_VALUE = "fournos-dashboard" +PROJECT_LABEL = "fournos-launcher/project" + + +def list_managed_cronjobs(namespace: str | None = None) -> list[dict]: + """List CronJobs managed by the dashboard.""" + _ensure_loaded() + if _batch_api is None: + return [] + ns = namespace or settings.fournos_namespace + try: + result = _batch_api.list_namespaced_cron_job( + namespace=ns, + label_selector=f"{SCHEDULE_LABEL}={SCHEDULE_LABEL_VALUE}", + ) + return [_cronjob_to_dict(cj) for cj in result.items] + except ApiException as exc: + logger.error("Failed to list CronJobs: %s", exc.reason) + return [] + + +def get_managed_cronjob(name: str, namespace: str | None = None) -> dict | None: + """Get a specific managed CronJob.""" + _ensure_loaded() + if _batch_api is None: + return None + ns = namespace or settings.fournos_namespace + try: + cj = _batch_api.read_namespaced_cron_job(name=name, namespace=ns) + return _cronjob_to_dict(cj) + except ApiException as exc: + if exc.status == 404: + return None + logger.error("Failed to get CronJob %s: %s", name, exc.reason) + return None + + +def create_cronjob( + name: str, + schedule: str, + project: str, + cluster: str, + pipeline: str, + preset: str, + image: str, + owner: str = "", + config_overrides: dict | None = None, + resolver_script: str = "", + resolver_image: str = "", + resolver_filename: str = "", + namespace: str | None = None, +) -> dict: + """Create a K8s CronJob that submits a FournosJob on schedule. + + If *resolver_script* is provided, the CronJob pod gets an init container + that runs the script and writes KEY=VALUE pairs to /shared/resolved.env. + The main submit container reads those values and injects them as + configOverrides into the FournosJob spec before submission. + + The file extension of *resolver_filename* determines the interpreter: + .py -> python, .sh (or default) -> sh -c. + """ + _ensure_loaded() + if _batch_api is None: + raise RuntimeError("Kubernetes client not available") + ns = namespace or settings.fournos_namespace + + fjob_spec = { + "apiVersion": f"{settings.fournos_api_group}/{settings.fournos_api_version}", + "kind": "FournosJob", + "metadata": { + "generateName": f"forge-{project.replace('_', '-')}-sched-", + "namespace": ns, + "labels": { + "fournos-launcher/schedule-name": name, + "fournos-launcher/trigger-type": "scheduled", + }, + }, + "spec": { + "cluster": cluster, + "displayName": f"{project} {preset}".strip(), + "owner": owner or "fournos-dashboard/scheduler", + "pipeline": pipeline, + "exclusive": True, + "executionEngine": { + "forge": { + "project": project, + "args": [preset] if preset else [], + "configOverrides": config_overrides or {}, + } + }, + }, + } + fjob_json = json.dumps(fjob_spec) + api_path = ( + f"/apis/{settings.fournos_api_group}/{settings.fournos_api_version}" + f"/namespaces/{ns}/{settings.fournos_job_plural}" + ) + + submit_common = ( + "token = open('/var/run/secrets/kubernetes.io/serviceaccount/token').read()\n" + "ctx = ssl.create_default_context(cafile='/var/run/secrets/kubernetes.io/serviceaccount/ca.crt')\n" + f"url = 'https://kubernetes.default.svc{api_path}'\n" + "print('Submitting FournosJob to', url)\n" + "print('Body:', json.dumps(body, indent=2))\n" + "data = json.dumps(body).encode()\n" + "req = urllib.request.Request(url, data=data, method='POST',\n" + " headers={'Authorization': 'Bearer ' + token, 'Content-Type': 'application/json'})\n" + "try:\n" + " resp = urllib.request.urlopen(req, context=ctx)\n" + " result = json.loads(resp.read())\n" + " print('FournosJob submitted:', result['metadata'].get('name', 'unknown'))\n" + "except urllib.error.HTTPError as e:\n" + " print('API error:', e.code, e.reason)\n" + " print(e.read().decode())\n" + " raise\n" + ) + + trigger_override = ( + "trigger = os.environ.get('FOURNOS_TRIGGER_TYPE', 'scheduled')\n" + "body['metadata'].setdefault('labels', {})['fournos-launcher/trigger-type'] = trigger\n" + ) + + if resolver_script: + submit_script = ( + "import json, os, ssl, urllib.request, urllib.error\n" + "body = json.loads(os.environ['FJOB_JSON'])\n" + + trigger_override + + "overrides = body['spec']['executionEngine']['forge'].setdefault('configOverrides', {})\n" + "env_file = '/shared/resolved.env'\n" + "if os.path.exists(env_file):\n" + " with open(env_file) as f:\n" + " for line in f:\n" + " line = line.strip()\n" + " if '=' in line and not line.startswith('#'):\n" + " k, v = line.split('=', 1)\n" + " overrides[k.strip()] = v.strip()\n" + " print(f'Resolved: {k.strip()} = {v.strip()}')\n" + + submit_common + ) + else: + submit_script = ( + "import json, os, ssl, urllib.request, urllib.error\n" + "body = json.loads(os.environ['FJOB_JSON'])\n" + + trigger_override + + submit_common + ) + + submit_container = client.V1Container( + name="submit", + image=image or "python:3.12-slim", + command=["python", "-c", submit_script], + env=[client.V1EnvVar(name="FJOB_JSON", value=fjob_json)], + ) + + init_containers = None + volumes = None + + if resolver_script: + is_python = resolver_filename.lower().endswith(".py") + default_resolver_img = "python:3.12-slim" if is_python else "alpine:latest" + script_key = resolver_filename or ("resolver.py" if is_python else "resolver.sh") + configmap_name = f"{name}-resolver" + + _create_resolver_configmap(configmap_name, script_key, resolver_script, ns) + + script_vol = client.V1Volume( + name="resolver-script", + config_map=client.V1ConfigMapVolumeSource( + name=configmap_name, + default_mode=0o755, + ), + ) + shared_vol = client.V1Volume( + name="shared", + empty_dir=client.V1EmptyDirVolumeSource(), + ) + volumes = [script_vol, shared_vol] + script_mount = client.V1VolumeMount( + name="resolver-script", mount_path="/resolver", read_only=True, + ) + shared_mount = client.V1VolumeMount(name="shared", mount_path="/shared") + submit_container.volume_mounts = [shared_mount] + + if is_python: + resolver_cmd = ["python", f"/resolver/{script_key}"] + else: + resolver_cmd = ["sh", f"/resolver/{script_key}"] + + init_containers = [ + client.V1Container( + name="resolver", + image=resolver_image or default_resolver_img, + command=resolver_cmd, + volume_mounts=[script_mount, shared_mount], + ) + ] + + annotations = { + "fournos-launcher/project": project, + "fournos-launcher/cluster": cluster, + "fournos-launcher/pipeline": pipeline, + "fournos-launcher/preset": preset, + "fournos-launcher/owner": owner, + } + if resolver_script: + annotations["fournos-launcher/resolver-configmap"] = configmap_name + annotations["fournos-launcher/resolver-filename"] = script_key + if resolver_image: + annotations["fournos-launcher/resolver-image"] = resolver_image + + cj_body = client.V1CronJob( + api_version="batch/v1", + kind="CronJob", + metadata=client.V1ObjectMeta( + name=name, + namespace=ns, + labels={ + SCHEDULE_LABEL: SCHEDULE_LABEL_VALUE, + PROJECT_LABEL: project, + }, + annotations=annotations, + ), + spec=client.V1CronJobSpec( + schedule=schedule, + suspend=False, + successful_jobs_history_limit=3, + failed_jobs_history_limit=3, + job_template=client.V1JobTemplateSpec( + spec=client.V1JobSpec( + template=client.V1PodTemplateSpec( + spec=client.V1PodSpec( + init_containers=init_containers, + containers=[submit_container], + volumes=volumes, + service_account_name="fournos-dashboard-sa", + restart_policy="Never", + ) + ), + backoff_limit=0, + ) + ), + ), + ) + + result = _batch_api.create_namespaced_cron_job(namespace=ns, body=cj_body) + return _cronjob_to_dict(result) + + +def _create_resolver_configmap( + cm_name: str, script_key: str, script_content: str, namespace: str +) -> None: + """Create a ConfigMap to hold a resolver script file.""" + _ensure_loaded() + if _core_api is None: + raise RuntimeError("Kubernetes client not available") + cm = client.V1ConfigMap( + api_version="v1", + kind="ConfigMap", + metadata=client.V1ObjectMeta( + name=cm_name, + namespace=namespace, + labels={ + SCHEDULE_LABEL: SCHEDULE_LABEL_VALUE, + "fournos-launcher/type": "resolver-script", + }, + ), + data={script_key: script_content}, + ) + try: + _core_api.create_namespaced_config_map(namespace=namespace, body=cm) + except ApiException as exc: + if exc.status == 409: + _core_api.replace_namespaced_config_map( + name=cm_name, namespace=namespace, body=cm, + ) + else: + raise + + +def get_resolver_script(configmap_name: str, namespace: str | None = None) -> tuple[str, str]: + """Read the resolver script from its ConfigMap. Returns (filename, content).""" + _ensure_loaded() + if _core_api is None: + return "", "" + ns = namespace or settings.fournos_namespace + try: + cm = _core_api.read_namespaced_config_map(name=configmap_name, namespace=ns) + if cm.data: + for key, value in cm.data.items(): + return key, value + except ApiException: + pass + return "", "" + + +def _delete_resolver_configmap(cm_name: str, namespace: str) -> None: + """Delete a resolver ConfigMap (best-effort).""" + _ensure_loaded() + if _core_api is None: + return + try: + _core_api.delete_namespaced_config_map(name=cm_name, namespace=namespace) + except ApiException: + pass + + +def trigger_cronjob(name: str, namespace: str | None = None) -> str: + """Manually trigger a CronJob by creating a one-off Job from its spec.""" + _ensure_loaded() + if _batch_api is None: + raise RuntimeError("Kubernetes client not available") + ns = namespace or settings.fournos_namespace + + cj = _batch_api.read_namespaced_cron_job(name=name, namespace=ns) + ts = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S") + job_name = f"{name}-manual-{ts}"[:63] + + job_spec = cj.spec.job_template.spec + + trigger_env = client.V1EnvVar(name="FOURNOS_TRIGGER_TYPE", value="manual") + if job_spec.template and job_spec.template.spec and job_spec.template.spec.containers: + for c in job_spec.template.spec.containers: + if c.env is None: + c.env = [] + c.env.append(trigger_env) + + job_body = client.V1Job( + api_version="batch/v1", + kind="Job", + metadata=client.V1ObjectMeta( + name=job_name, + namespace=ns, + labels={ + SCHEDULE_LABEL: SCHEDULE_LABEL_VALUE, + "fournos-launcher/triggered-by": "manual", + }, + annotations={"cronjob.kubernetes.io/instantiate": "manual"}, + ), + spec=job_spec, + ) + + _batch_api.create_namespaced_job(namespace=ns, body=job_body) + return job_name + + +def delete_cronjob(name: str, namespace: str | None = None) -> None: + """Delete a managed CronJob and its resolver ConfigMap if present.""" + _ensure_loaded() + if _batch_api is None: + raise RuntimeError("Kubernetes client not available") + ns = namespace or settings.fournos_namespace + cj = get_managed_cronjob(name, ns) + _batch_api.delete_namespaced_cron_job( + name=name, + namespace=ns, + propagation_policy="Foreground", + ) + if cj and cj.get("resolver_configmap"): + _delete_resolver_configmap(cj["resolver_configmap"], ns) + + +def patch_cronjob_suspend( + name: str, suspend: bool, namespace: str | None = None +) -> dict: + """Pause or resume a CronJob by toggling spec.suspend.""" + _ensure_loaded() + if _batch_api is None: + raise RuntimeError("Kubernetes client not available") + ns = namespace or settings.fournos_namespace + result = _batch_api.patch_namespaced_cron_job( + name=name, + namespace=ns, + body={"spec": {"suspend": suspend}}, + ) + return _cronjob_to_dict(result) + + +def _cronjob_to_dict(cj: Any) -> dict: + """Serialise a V1CronJob into a plain dict for templates.""" + meta = cj.metadata + annotations = meta.annotations or {} + resolver_configmap = annotations.get("fournos-launcher/resolver-configmap", "") + resolver_image = annotations.get("fournos-launcher/resolver-image", "") + resolver_filename = annotations.get("fournos-launcher/resolver-filename", "") + return { + "name": meta.name, + "namespace": meta.namespace, + "schedule": cj.spec.schedule, + "suspend": cj.spec.suspend or False, + "project": annotations.get("fournos-launcher/project", ""), + "cluster": annotations.get("fournos-launcher/cluster", ""), + "pipeline": annotations.get("fournos-launcher/pipeline", ""), + "preset": annotations.get("fournos-launcher/preset", ""), + "owner": annotations.get("fournos-launcher/owner", ""), + "resolver_configmap": resolver_configmap, + "resolver_image": resolver_image, + "resolver_filename": resolver_filename, + "has_resolver": bool(resolver_configmap), + "created_at": meta.creation_timestamp.isoformat() if meta.creation_timestamp else "", + "last_schedule": ( + cj.status.last_schedule_time.isoformat() + if cj.status and cj.status.last_schedule_time + else "" + ), + "active_count": len(cj.status.active) if cj.status and cj.status.active else 0, + } diff --git a/fournos-ui/app/main.py b/fournos-ui/app/main.py new file mode 100644 index 0000000..72e51a0 --- /dev/null +++ b/fournos-ui/app/main.py @@ -0,0 +1,930 @@ +"""Fournos Launcher Dashboard -- production FastAPI application.""" + +from __future__ import annotations + +import asyncio +import logging +import re +from contextlib import asynccontextmanager +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +from fastapi import FastAPI, Form, HTTPException, Query, Request +from fastapi.responses import HTMLResponse, RedirectResponse, StreamingResponse +from fastapi.staticfiles import StaticFiles +from jinja2 import Environment, FileSystemLoader + +from app import db, k8s_client, watcher +from app.config import settings +from app.forge_discovery import discover_projects, get_project_presets + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Lifespan +# --------------------------------------------------------------------------- + +@asynccontextmanager +async def lifespan(app: FastAPI): + logging.basicConfig(level=getattr(logging, settings.log_level)) + await db.init_db() + watcher.start_watcher() + yield + +app = FastAPI(title="Fournos Launcher Dashboard", lifespan=lifespan) + +BASE_DIR = Path(__file__).resolve().parent +app.mount("/static", StaticFiles(directory=BASE_DIR / "static"), name="static") + +_jinja_env = Environment( + loader=FileSystemLoader(str(BASE_DIR / "templates")), + autoescape=True, + auto_reload=False, +) + +# --------------------------------------------------------------------------- +# Template helpers +# --------------------------------------------------------------------------- + +def _format_age(timestamp_str: str) -> str: + from dateutil.parser import parse + + try: + created = parse(timestamp_str) + except Exception: + return "?" + delta = datetime.now(timezone.utc) - created + total_seconds = int(delta.total_seconds()) + if total_seconds < 0: + return "0s" + if total_seconds < 60: + return f"{total_seconds}s" + if total_seconds < 3600: + return f"{total_seconds // 60}m" + hours = total_seconds // 3600 + mins = (total_seconds % 3600) // 60 + if hours < 24: + return f"{hours}h {mins}m" + days = hours // 24 + return f"{days}d {hours % 24}h" + + +def _format_duration(seconds: float | None) -> str: + if seconds is None: + return "-" + s = int(seconds) + h, remainder = divmod(s, 3600) + m, sec = divmod(remainder, 60) + return f"{h:02d}h {m:02d}m {sec:02d}s" + + +def _phase_class(phase: str) -> str: + return { + "Running": "phase-running", + "Succeeded": "phase-succeeded", + "Failed": "phase-failed", + "Stopped": "phase-stopped", + "Resolving": "phase-resolving", + "Pending": "phase-resolving", + }.get(phase, "phase-unknown") + + +def _extract_forge_info(job: dict) -> dict: + forge = job.get("spec", {}).get("executionEngine", {}).get("forge", {}) + env = job.get("spec", {}).get("env", {}) + pr_number = env.get("PULL_NUMBER", "") + pr_title = env.get("PULL_TITLE", "") + repo_owner = env.get("REPO_OWNER", "") + repo_name = env.get("REPO_NAME", "") + pr_url = f"https://github.com/{repo_owner}/{repo_name}/pull/{pr_number}" if pr_number else "" + return { + "project": forge.get("project", ""), + "args": forge.get("args", []), + "config_overrides": forge.get("configOverrides", {}), + "pr_number": pr_number, + "pr_title": pr_title, + "pr_url": pr_url, + } + + +def _parse_task_progress(message: str) -> dict | None: + m = re.search( + r"Tasks Completed:\s*(\d+)\s*\(Failed:\s*(\d+),\s*Cancelled\s*(\d+)\),\s*Incomplete:\s*(\d+),\s*Skipped:\s*(\d+)", + message, + ) + if not m: + return None + return { + "completed": int(m.group(1)), + "failed": int(m.group(2)), + "cancelled": int(m.group(3)), + "incomplete": int(m.group(4)), + "skipped": int(m.group(5)), + "total": int(m.group(1)) + int(m.group(4)) + int(m.group(5)), + } + + +def _build_timeline(stages: list[dict]) -> list[dict]: + from dateutil.parser import parse + + now = datetime.now(timezone.utc) + n = len(stages) or 1 + equal_pct = 100.0 / n + + result = [] + for s in stages: + start = parse(s["startTime"]) if s["startTime"] else None + end = parse(s["completionTime"]) if s["completionTime"] else None + if start and end: + dur = (end - start).total_seconds() + elif start: + dur = (now - start).total_seconds() + else: + dur = 0 + dur = max(dur, 0) + + if dur < 60: + dur_label = f"{int(dur)}s" + elif dur < 3600: + dur_label = f"{int(dur // 60)}m {int(dur % 60)}s" + else: + dur_label = f"{int(dur // 3600)}h {int((dur % 3600) // 60)}m" + + status_class = { + "Succeeded": "ptl-ok", + "Running": "ptl-run", + "Failed": "ptl-err", + "Pending": "ptl-wait", + "Cancelled": "ptl-cancel", + "Skipped": "ptl-skip", + }.get(s["status"], "ptl-wait") + + result.append({ + **s, + "width_pct": equal_pct, + "min_width": 8, + "duration_label": dur_label if s["startTime"] else "", + "status_class": status_class, + }) + return result + + +def _extract_mlflow_url(status: dict) -> str: + """Extract MLflow run URL from FournosJob status.""" + mlflow = ( + status.get("engineStatus", {}) + .get("forge", {}) + .get("exportArtifacts", {}) + .get("caliper_artifacts_export", {}) + .get("backends", {}) + .get("mlflow", {}) + ) + return mlflow.get("run_url", "") if mlflow else "" + + +_CACHE_BUST = str(int(datetime.now(timezone.utc).timestamp())) + +_jinja_env.globals.update( + format_age=_format_age, + format_duration=_format_duration, + phase_class=_phase_class, + extract_forge_info=_extract_forge_info, + parse_task_progress=_parse_task_progress, + build_timeline=_build_timeline, + extract_mlflow_url=_extract_mlflow_url, + url_for=lambda name, **kw: app.url_path_for(name, **kw), + cache_bust=_CACHE_BUST, +) + + +_NAV_MAP = { + "jobs_list.html": "jobs", + "job_detail.html": "jobs", + "components/jobs_table_body.html": "jobs", + "submit_job.html": "submit", + "schedules.html": "schedules", + "schedule_runs.html": "schedules", +} + + +def _render(template_name: str, **context: Any) -> HTMLResponse: + context.setdefault("active_nav", _NAV_MAP.get(template_name, "")) + tpl = _jinja_env.get_template(template_name) + return HTMLResponse(tpl.render(**context)) + + +# --------------------------------------------------------------------------- +# Data fetching helpers +# --------------------------------------------------------------------------- + +_COMPLETED_GRACE_SECONDS = 180 # keep completed jobs on Live tab for 3 minutes + + +def _get_live_jobs_sync() -> list[dict]: + """Get FournosJobs from K8s, sorted newest-first, hiding old completed jobs.""" + from dateutil.parser import parse + + jobs = k8s_client.list_fournos_jobs() + now = datetime.now(timezone.utc) + visible: list[dict] = [] + for j in jobs: + phase = j.get("status", {}).get("phase", "") + if phase in ("Succeeded", "Failed", "Stopped"): + conditions = j.get("status", {}).get("conditions", []) + last_ts = None + for c in conditions: + ts_str = c.get("lastTransitionTime") + if ts_str: + try: + last_ts = parse(ts_str) + except Exception: + pass + if last_ts and (now - last_ts).total_seconds() > _COMPLETED_GRACE_SECONDS: + continue + visible.append(j) + + visible.sort( + key=lambda j: j.get("metadata", {}).get("creationTimestamp", ""), + reverse=True, + ) + return visible + + +async def _get_live_jobs() -> list[dict]: + """Async wrapper -- offloads blocking K8s I/O to a thread.""" + return await asyncio.to_thread(_get_live_jobs_sync) + + +def _compute_current_steps_sync(jobs: list[dict]) -> dict[str, dict]: + """For each running job, fetch the currently active pipeline step.""" + steps: dict[str, dict] = {} + for j in jobs: + phase = j.get("status", {}).get("phase", "") + if phase not in ("Running", "Admitted"): + continue + name = j.get("metadata", {}).get("name", "") + try: + step = k8s_client.get_current_step_for_job(name) + if step: + steps[name] = step + except Exception: + pass + return steps + + +async def _compute_current_steps(jobs: list[dict]) -> dict[str, dict]: + """Async wrapper -- offloads blocking K8s I/O to a thread.""" + return await asyncio.to_thread(_compute_current_steps_sync, jobs) + + +def _get_pipeline_stages_sync(job: dict) -> list[dict]: + """Get pipeline stages for a job from its PipelineRun.""" + job_name = job.get("metadata", {}).get("name", "") + + pr_name = job.get("status", {}).get("pipelineRun", "") + if pr_name: + pr = k8s_client.get_pipelinerun(pr_name) + if pr: + return k8s_client.extract_pipeline_stages(pr) + + prs = k8s_client.list_pipelineruns_for_job(job_name) + if prs: + return k8s_client.extract_pipeline_stages(prs[0]) + + return [] + + +async def _get_pipeline_stages(job: dict) -> list[dict]: + """Async wrapper -- offloads blocking K8s I/O to a thread.""" + return await asyncio.to_thread(_get_pipeline_stages_sync, job) + + +# --------------------------------------------------------------------------- +# Routes: Jobs +# --------------------------------------------------------------------------- + +@app.get("/", response_class=HTMLResponse) +async def jobs_list( + request: Request, + tab: str = Query("live", pattern="^(live|history)$"), + project: str = Query("", alias="project"), + cluster: str = Query("", alias="cluster"), + status: str = Query("", alias="status"), + owner: str = Query("", alias="owner"), + page: int = Query(1, ge=1), +): + per_page = 50 + filters = {"project": project, "cluster": cluster, "status": status, "owner": owner} + + if tab == "live": + jobs = await _get_live_jobs() + if project: + jobs = [j for j in jobs if _extract_forge_info(j).get("project") == project] + if cluster: + jobs = [j for j in jobs if j.get("spec", {}).get("cluster") == cluster] + if status: + jobs = [j for j in jobs if j.get("status", {}).get("phase") == status] + if owner: + jobs = [j for j in jobs if j.get("spec", {}).get("owner") == owner] + total = len(jobs) + all_clusters = _collect_clusters(jobs) + offset = (page - 1) * per_page + jobs = jobs[offset:offset + per_page] + history_jobs = [] + total_history = 0 + else: + jobs = [] + all_clusters = [] + async with db.async_session() as session: + history_jobs_db, total_history = await db.list_jobs( + session, + project=project or None, + cluster=cluster or None, + status=status or None, + owner=owner or None, + limit=per_page, + offset=(page - 1) * per_page, + ) + history_jobs = [_db_job_to_dict(j) for j in history_jobs_db] + total = total_history + + projects_list = [p.name for p in discover_projects()] + clusters = all_clusters + current_steps = await _compute_current_steps(jobs) if tab == "live" else {} + + return _render( + "jobs_list.html", + jobs=jobs, + history_jobs=history_jobs, + tab=tab, + filters=filters, + projects=projects_list, + clusters=clusters, + page=page, + per_page=per_page, + total=total, + current_steps=current_steps, + ) + + +@app.get("/api/jobs-table", response_class=HTMLResponse) +async def jobs_table_partial( + request: Request, + project: str = Query(""), + cluster: str = Query(""), + status: str = Query(""), + owner: str = Query(""), +): + jobs = await _get_live_jobs() + if project: + jobs = [j for j in jobs if _extract_forge_info(j).get("project") == project] + if cluster: + jobs = [j for j in jobs if j.get("spec", {}).get("cluster") == cluster] + if status: + jobs = [j for j in jobs if j.get("status", {}).get("phase") == status] + if owner: + jobs = [j for j in jobs if j.get("spec", {}).get("owner") == owner] + current_steps = await _compute_current_steps(jobs) + return _render("components/jobs_table_body.html", jobs=jobs, current_steps=current_steps) + + +@app.get("/jobs/{job_name}", response_class=HTMLResponse) +async def job_detail(request: Request, job_name: str): + job = None + source = "live" + pods: list[dict] = [] + stages: list[dict] = [] + + job = await asyncio.to_thread(k8s_client.get_fournos_job, job_name) + if job: + pods = await asyncio.to_thread(k8s_client.list_pods_for_job, job_name) + stages = await _get_pipeline_stages(job) + + if not job: + source = "history" + async with db.async_session() as session: + db_job = await db.get_job_by_name(session, job_name) + if db_job is None: + raise HTTPException(status_code=404, detail="Job not found") + + job = _db_job_to_fjob_dict(db_job) + + return _render( + "job_detail.html", + job=job, + pods=pods, + stages=stages, + source=source, + ) + + +@app.get("/api/jobs/{job_name}/detail-partial", response_class=HTMLResponse) +async def job_detail_partial(request: Request, job_name: str): + """Return the dynamic portions of the job detail page for HTMX polling.""" + job = await asyncio.to_thread(k8s_client.get_fournos_job, job_name) + if not job: + return HTMLResponse("") + pods = await asyncio.to_thread(k8s_client.list_pods_for_job, job_name) + stages = await _get_pipeline_stages(job) + return _render( + "components/job_detail_dynamic.html", + job=job, + pods=pods, + stages=stages, + ) + + +@app.post("/api/jobs/{job_name}/cancel") +async def cancel_job(job_name: str): + try: + await asyncio.to_thread(k8s_client.shutdown_fournos_job, job_name) + return {"status": "ok", "message": f"Shutdown requested for {job_name}"} + except Exception as exc: + raise HTTPException(status_code=500, detail=str(exc)) + + +@app.post("/api/jobs/{job_name}/rerun") +async def rerun_job(job_name: str): + """Clone an existing FournosJob's spec into a brand-new job.""" + job = await _get_job_for_rerun(job_name) + if job is None: + raise HTTPException(status_code=404, detail="Job not found") + + spec = dict(job.get("spec", {})) + forge = spec.get("executionEngine", {}).get("forge", {}) + project = forge.get("project", "unknown") + + spec.pop("shutdown", None) + + new_name = sanitize_job_name(f"forge-{project}") + body = { + "apiVersion": f"{settings.fournos_api_group}/{settings.fournos_api_version}", + "kind": "FournosJob", + "metadata": { + "name": new_name, + "namespace": settings.fournos_namespace, + }, + "spec": spec, + } + + try: + created = await asyncio.to_thread(k8s_client.create_fournos_job, body) + created_name = created.get("metadata", {}).get("name", new_name) + return {"status": "ok", "job_name": created_name, "redirect": f"/jobs/{created_name}"} + except Exception as exc: + raise HTTPException(status_code=500, detail=str(exc)) + + +async def _get_job_for_rerun(job_name: str) -> dict | None: + """Fetch a FournosJob by name from live K8s or DB history.""" + live = await asyncio.to_thread(k8s_client.get_fournos_job, job_name) + if live: + return live + async with db.async_session() as session: + db_job = await db.get_job_by_name(session, job_name) + if db_job: + return _db_job_to_fjob_dict(db_job) + return None + + +@app.delete("/api/history/{job_name}") +async def delete_history_job(job_name: str): + """Delete a job from the history database.""" + async with db.async_session() as session: + async with session.begin(): + deleted = await db.delete_job_by_name(session, job_name) + if not deleted: + raise HTTPException(status_code=404, detail="Job not found in history") + return {"status": "ok"} + + +@app.get("/api/jobs/{job_name}/logs/{pod_name}") +async def stream_logs(job_name: str, pod_name: str): + """Stream live pod logs via SSE (only for running jobs).""" + job_pods = await asyncio.to_thread(k8s_client.list_pods_for_job, job_name) + pod_names = {p["name"] for p in job_pods} + if pod_name not in pod_names: + raise HTTPException(status_code=404, detail="Pod not found for this job") + + async def generate(): + stop = asyncio.Event() + queue: asyncio.Queue[str | None] = asyncio.Queue(maxsize=64) + loop = asyncio.get_event_loop() + + def _reader(): + try: + for line in k8s_client.read_pod_log(pod_name, follow=True): + if stop.is_set(): + break + try: + loop.call_soon_threadsafe(queue.put_nowait, line) + except asyncio.QueueFull: + pass + finally: + loop.call_soon_threadsafe(queue.put_nowait, None) + + task = asyncio.get_event_loop().run_in_executor(None, _reader) + try: + while True: + line = await queue.get() + if line is None: + break + yield f"data: {line}\n\n" + finally: + stop.set() + + return StreamingResponse(generate(), media_type="text/event-stream") + + + + +# --------------------------------------------------------------------------- +# Routes: Submit Job +# --------------------------------------------------------------------------- + +@app.get("/submit", response_class=HTMLResponse) +async def submit_form(request: Request): + projects = discover_projects() + return _render( + "submit_job.html", + projects=projects, + pipelines=list(settings.default_pipelines), + ) + + +@app.get("/api/project-info/{project_name}") +async def project_info_api(project_name: str): + from app.forge_discovery import get_project + proj = get_project(project_name) + if proj is None: + return {"presets": [], "cluster": ""} + return {"presets": proj.presets, "cluster": proj.cluster} + + +def _fetch_github_open_prs() -> list[dict]: + """Blocking call to the GitHub API -- run via asyncio.to_thread.""" + import urllib.request + import json as _json + + url = f"https://api.github.com/repos/{settings.forge_github_repo}/pulls?state=open&per_page=100" + req = urllib.request.Request(url, headers={"Accept": "application/vnd.github+json"}) + with urllib.request.urlopen(req, timeout=10) as resp: + prs = _json.loads(resp.read()) + + return [ + { + "number": pr["number"], + "title": pr["title"], + "author": pr["user"]["login"], + "head_sha": pr["head"]["sha"], + "branch": pr["head"]["ref"], + "draft": pr["draft"], + } + for pr in prs + ] + + +@app.get("/api/github/open-prs") +async def github_open_prs(): + """Fetch open pull requests from the Forge GitHub repo (public, no token needed).""" + try: + return await asyncio.to_thread(_fetch_github_open_prs) + except Exception as exc: + raise HTTPException(status_code=502, detail=f"GitHub API error: {exc}") + + +@app.post("/submit") +async def submit_job( + request: Request, + project: str = Form(...), + cluster: str = Form(...), + pipeline: str = Form("forge-test-only"), + preset: str = Form(""), + version: str = Form(""), + owner: str = Form(""), + exclusive: str = Form("false"), + config_overrides_raw: str = Form(""), + pull_sha: str = Form(""), +): + exclusive_bool = exclusive.lower() in ("true", "on", "1", "yes") + + config_overrides: dict[str, Any] = {} + if config_overrides_raw.strip(): + for line in config_overrides_raw.strip().splitlines(): + line = line.strip() + if ":" in line: + k, v = line.split(":", 1) + config_overrides[k.strip()] = v.strip() + + if version: + version_key = _get_version_config_key(project) + config_overrides[version_key] = version + + args = [preset] if preset else [] + + job_name = sanitize_job_name(f"forge-{project}") + + pull_sha = pull_sha.strip() + env: dict[str, str] = {} + if pull_sha: + env["PULL_PULL_SHA"] = pull_sha + + body = { + "apiVersion": f"{settings.fournos_api_group}/{settings.fournos_api_version}", + "kind": "FournosJob", + "metadata": { + "name": job_name, + "namespace": settings.fournos_namespace, + }, + "spec": { + "cluster": cluster, + "displayName": f"{project} {preset}".strip(), + "owner": owner or "fournos-dashboard", + "pipeline": pipeline, + "exclusive": exclusive_bool, + "executionEngine": { + "forge": { + "project": project, + "args": args, + "configOverrides": config_overrides, + } + }, + }, + } + + if env: + body["spec"]["env"] = env + + try: + created = await asyncio.to_thread(k8s_client.create_fournos_job, body) + except Exception as exc: + projects = discover_projects() + return _render( + "submit_job.html", + projects=projects, + pipelines=list(settings.default_pipelines), + error=str(exc), + ) + + created_name = created.get("metadata", {}).get("name", job_name) + + try: + async with db.async_session() as session: + async with session.begin(): + await db.upsert_job( + session, + name=created_name, + project=project, + preset=preset, + cluster=cluster, + pipeline=pipeline, + owner=owner or "fournos-dashboard", + status="Pending", + config_overrides=config_overrides, + fjob_spec=body.get("spec", {}), + ) + except Exception as exc: + logger.error("DB upsert failed for job %s (job was created in K8s): %s", created_name, exc) + + return RedirectResponse(url=f"/jobs/{created_name}", status_code=303) + + +# --------------------------------------------------------------------------- +# Routes: Schedules +# --------------------------------------------------------------------------- + +@app.get("/schedules", response_class=HTMLResponse) +async def schedules_list(request: Request): + cronjobs = await asyncio.to_thread(k8s_client.list_managed_cronjobs) + projects = discover_projects() + return _render( + "schedules.html", + cronjobs=cronjobs, + projects=projects, + pipelines=list(settings.default_pipelines), + ) + + +@app.get("/schedules/{name}/runs", response_class=HTMLResponse) +async def schedule_runs(request: Request, name: str): + """Show all jobs triggered by a specific schedule.""" + async with db.async_session() as session: + jobs = await db.list_jobs_by_schedule(session, name) + runs = [] + for j in jobs: + runs.append({ + "name": j.name, + "status": j.status, + "preset": j.preset, + "trigger_type": j.trigger_type or "scheduled", + "duration_seconds": j.duration_seconds, + "mlflow_url": j.mlflow_url, + "created_at": j.created_at.isoformat() if j.created_at else "", + }) + return _render("schedule_runs.html", schedule_name=name, runs=runs) + + +@app.post("/schedules") # handles both create and edit +async def create_schedule( + request: Request, + name: str = Form(...), + project: str = Form(...), + cluster: str = Form(...), + pipeline: str = Form("forge-test-only"), + preset: str = Form(""), + cron_expr: str = Form(...), + image_source: str = Form(""), + owner: str = Form(""), + resolver_script: str = Form(""), + resolver_image: str = Form(""), + resolver_filename: str = Form(""), + edit_target: str = Form(""), +): + try: + if edit_target and edit_target != name: + await asyncio.to_thread( + k8s_client.create_cronjob, + name=name, + schedule=cron_expr, + project=project, + cluster=cluster, + pipeline=pipeline, + preset=preset, + image=image_source, + owner=owner, + resolver_script=resolver_script.strip().replace("\r\n", "\n").replace("\r", "\n"), + resolver_image=resolver_image.strip(), + resolver_filename=resolver_filename.strip(), + ) + try: + await asyncio.to_thread(k8s_client.delete_cronjob, edit_target) + except Exception as del_exc: + logger.warning("Failed to delete old schedule %s after replacement: %s", edit_target, del_exc) + else: + if edit_target: + await asyncio.to_thread(k8s_client.delete_cronjob, edit_target) + await asyncio.to_thread( + k8s_client.create_cronjob, + name=name, + schedule=cron_expr, + project=project, + cluster=cluster, + pipeline=pipeline, + preset=preset, + image=image_source, + owner=owner, + resolver_script=resolver_script.strip().replace("\r\n", "\n").replace("\r", "\n"), + resolver_image=resolver_image.strip(), + resolver_filename=resolver_filename.strip(), + ) + + return RedirectResponse(url="/schedules", status_code=303) + except Exception as exc: + cronjobs = await asyncio.to_thread(k8s_client.list_managed_cronjobs) + projects = discover_projects() + return _render( + "schedules.html", + cronjobs=cronjobs, + projects=projects, + pipelines=list(settings.default_pipelines), + error=str(exc), + ) + + +@app.post("/api/schedules/{name}/toggle") +async def toggle_schedule(name: str): + cj = await asyncio.to_thread(k8s_client.get_managed_cronjob, name) + if cj is None: + raise HTTPException(404, "Schedule not found") + await asyncio.to_thread(k8s_client.patch_cronjob_suspend, name, not cj["suspend"]) + return {"status": "ok"} + + +@app.get("/api/schedules/{name}/resolver") +async def get_resolver_script(name: str): + """Return the resolver script content for a schedule.""" + cj = await asyncio.to_thread(k8s_client.get_managed_cronjob, name) + if cj is None: + raise HTTPException(404, "Schedule not found") + cm_name = cj.get("resolver_configmap", "") + if not cm_name: + raise HTTPException(404, "No resolver script configured for this schedule") + filename, content = await asyncio.to_thread(k8s_client.get_resolver_script, cm_name) + if not content: + raise HTTPException(404, "Resolver ConfigMap not found") + return {"filename": filename, "content": content} + + +@app.post("/api/schedules/{name}/trigger") +async def trigger_schedule(name: str): + """Manually trigger a CronJob by creating a one-off Job from it.""" + try: + job = await asyncio.to_thread(k8s_client.trigger_cronjob, name) + return {"status": "ok", "job_name": job} + except Exception as exc: + raise HTTPException(500, str(exc)) + + +@app.post("/api/schedules/{name}/delete") +async def delete_schedule(name: str): + try: + await asyncio.to_thread(k8s_client.delete_cronjob, name) + return {"status": "ok"} + except Exception as exc: + raise HTTPException(500, str(exc)) + + +# --------------------------------------------------------------------------- +# Conversion helpers +# --------------------------------------------------------------------------- + +def _collect_clusters(live_jobs: list[dict]) -> list[str]: + """Collect unique cluster names from live jobs.""" + clusters = set() + for j in live_jobs: + c = j.get("spec", {}).get("cluster", "") + if c: + clusters.add(c) + return sorted(clusters) + + +def _db_job_to_dict(job: db.Job) -> dict: + """Convert a DB Job row to a dict suitable for the history table template.""" + return { + "name": job.name, + "project": job.project, + "preset": job.preset, + "cluster": job.cluster, + "pipeline": job.pipeline, + "owner": job.owner, + "phase": job.status, + "message": job.message, + "created_at": job.created_at.isoformat() if job.created_at else "", + "completed_at": job.completed_at.isoformat() if job.completed_at else "", + "duration_seconds": job.duration_seconds, + "mlflow_url": job.mlflow_url, + "error_message": job.error_message, + "triggered_by_schedule": job.triggered_by_schedule, + "trigger_type": job.trigger_type or "manual", + "source": "history", + } + + +def _db_job_to_fjob_dict(job: db.Job) -> dict: + """Convert a DB Job row to a FournosJob-like dict for the detail template.""" + spec = job.fjob_spec or {} + status = job.fjob_status or {} + + forge = spec.get("executionEngine", {}).get("forge", {}) + if not forge: + forge = {"project": job.project, "args": job.preset.split() if job.preset else [], "configOverrides": job.config_overrides or {}} + spec.setdefault("executionEngine", {})["forge"] = forge + + spec.setdefault("cluster", job.cluster) + spec.setdefault("pipeline", job.pipeline) + spec.setdefault("owner", job.owner) + spec.setdefault("displayName", f"{job.project} {job.preset}".strip()) + spec.setdefault("exclusive", True) + spec.setdefault("env", {}) + spec.setdefault("secretRefs", []) + + status.setdefault("phase", job.status) + status.setdefault("message", job.message) + status.setdefault("conditions", []) + + return { + "metadata": { + "name": job.name, + "namespace": settings.fournos_namespace, + "creationTimestamp": job.created_at.isoformat() if job.created_at else "", + "uid": job.id, + }, + "spec": spec, + "status": status, + "_source": "history", + "_duration_seconds": job.duration_seconds, + "_mlflow_url": job.mlflow_url, + "_ci_artifacts_url": job.ci_artifacts_url, + } + + +_PROJECT_VERSION_KEYS: dict[str, str] = { + "mcp_gateway": "infrastructure.mcp_gateway_version", +} + + +def _get_version_config_key(project: str) -> str: + """Return the configOverrides key used to pass the version for a project.""" + return _PROJECT_VERSION_KEYS.get(project, "infrastructure.version") + + +def sanitize_job_name(prefix: str) -> str: + """Generate a K8s-safe job name with timestamp.""" + ts = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S") + name = f"{prefix}-{ts}".lower() + name = re.sub(r"[^a-z0-9-]", "-", name) + name = re.sub(r"-+", "-", name).strip("-") + return name[:63] + + diff --git a/fournos-ui/app/models.py b/fournos-ui/app/models.py new file mode 100644 index 0000000..1bc83e9 --- /dev/null +++ b/fournos-ui/app/models.py @@ -0,0 +1,15 @@ +"""Pydantic models for the Fournos Dashboard.""" + +from __future__ import annotations + +from pydantic import BaseModel, Field + + +class ProjectInfo(BaseModel): + """Discovered Forge project metadata.""" + + name: str + cluster: str = "" + presets: list[str] = Field(default_factory=list) + config_keys: list[str] = Field(default_factory=list) + has_cli: bool = False diff --git a/fournos-ui/app/static/htmx-sse.js b/fournos-ui/app/static/htmx-sse.js new file mode 100644 index 0000000..462382d --- /dev/null +++ b/fournos-ui/app/static/htmx-sse.js @@ -0,0 +1,386 @@ +/* +Server Sent Events Extension +============================ +This extension adds support for Server Sent Events to htmx. See /www/extensions/sse.md for usage instructions. + +*/ + +(function() { + + /** @type {import("../htmx").HtmxInternalApi} */ + var api; + + htmx.defineExtension("sse", { + + /** + * Init saves the provided reference to the internal HTMX API. + * + * @param {import("../htmx").HtmxInternalApi} api + * @returns void + */ + init: function(apiRef) { + // store a reference to the internal API. + api = apiRef; + + // set a function in the public API for creating new EventSource objects + if (htmx.createEventSource == undefined) { + htmx.createEventSource = createEventSource; + } + }, + + /** + * onEvent handles all events passed to this extension. + * + * @param {string} name + * @param {Event} evt + * @returns void + */ + onEvent: function(name, evt) { + + var parent = evt.target || evt.detail.elt; + switch (name) { + + case "htmx:beforeCleanupElement": + var internalData = api.getInternalData(parent) + // Try to remove remove an EventSource when elements are removed + if (internalData.sseEventSource) { + internalData.sseEventSource.close(); + } + + return; + + // Try to create EventSources when elements are processed + case "htmx:afterProcessNode": + ensureEventSourceOnElement(parent); + } + } + }); + + /////////////////////////////////////////////// + // HELPER FUNCTIONS + /////////////////////////////////////////////// + + + /** + * createEventSource is the default method for creating new EventSource objects. + * it is hoisted into htmx.config.createEventSource to be overridden by the user, if needed. + * + * @param {string} url + * @returns EventSource + */ + function createEventSource(url) { + return new EventSource(url, { withCredentials: true }); + } + + function splitOnWhitespace(trigger) { + return trigger.trim().split(/\s+/); + } + + function getLegacySSEURL(elt) { + var legacySSEValue = api.getAttributeValue(elt, "hx-sse"); + if (legacySSEValue) { + var values = splitOnWhitespace(legacySSEValue); + for (var i = 0; i < values.length; i++) { + var value = values[i].split(/:(.+)/); + if (value[0] === "connect") { + return value[1]; + } + } + } + } + + function getLegacySSESwaps(elt) { + var legacySSEValue = api.getAttributeValue(elt, "hx-sse"); + var returnArr = []; + if (legacySSEValue != null) { + var values = splitOnWhitespace(legacySSEValue); + for (var i = 0; i < values.length; i++) { + var value = values[i].split(/:(.+)/); + if (value[0] === "swap") { + returnArr.push(value[1]); + } + } + } + return returnArr; + } + + /** + * registerSSE looks for attributes that can contain sse events, right + * now hx-trigger and sse-swap and adds listeners based on these attributes too + * the closest event source + * + * @param {HTMLElement} elt + */ + function registerSSE(elt) { + // Add message handlers for every `sse-swap` attribute + queryAttributeOnThisOrChildren(elt, "sse-swap").forEach(function (child) { + // Find closest existing event source + var sourceElement = api.getClosestMatch(child, hasEventSource); + if (sourceElement == null) { + // api.triggerErrorEvent(elt, "htmx:noSSESourceError") + return null; // no eventsource in parentage, orphaned element + } + + // Set internalData and source + var internalData = api.getInternalData(sourceElement); + var source = internalData.sseEventSource; + + var sseSwapAttr = api.getAttributeValue(child, "sse-swap"); + if (sseSwapAttr) { + var sseEventNames = sseSwapAttr.split(","); + } else { + var sseEventNames = getLegacySSESwaps(child); + } + + for (var i = 0; i < sseEventNames.length; i++) { + var sseEventName = sseEventNames[i].trim(); + var listener = function(event) { + + // If the source is missing then close SSE + if (maybeCloseSSESource(sourceElement)) { + return; + } + + // If the body no longer contains the element, remove the listener + if (!api.bodyContains(child)) { + source.removeEventListener(sseEventName, listener); + return; + } + + // swap the response into the DOM and trigger a notification + if(!api.triggerEvent(elt, "htmx:sseBeforeMessage", event)) { + return; + } + swap(child, event.data); + api.triggerEvent(elt, "htmx:sseMessage", event); + }; + + // Register the new listener + api.getInternalData(child).sseEventListener = listener; + source.addEventListener(sseEventName, listener); + } + }); + + // Add message handlers for every `hx-trigger="sse:*"` attribute + queryAttributeOnThisOrChildren(elt, "hx-trigger").forEach(function(child) { + // Find closest existing event source + var sourceElement = api.getClosestMatch(child, hasEventSource); + if (sourceElement == null) { + // api.triggerErrorEvent(elt, "htmx:noSSESourceError") + return null; // no eventsource in parentage, orphaned element + } + + // Set internalData and source + var internalData = api.getInternalData(sourceElement); + var source = internalData.sseEventSource; + + var sseEventName = api.getAttributeValue(child, "hx-trigger"); + if (sseEventName == null) { + return; + } + + // Only process hx-triggers for events with the "sse:" prefix + if (sseEventName.slice(0, 4) != "sse:") { + return; + } + + // remove the sse: prefix from here on out + sseEventName = sseEventName.substr(4); + + var listener = function(event) { + if (maybeCloseSSESource(sourceElement)) { + return; + } + + if (!api.bodyContains(child)) { + source.removeEventListener(sseEventName, listener); + return; + } + + api.triggerEvent(child, "htmx:sseMessage", event); + htmx.trigger(child, sseEventName, event); + }; + + api.getInternalData(child).sseEventListener = listener; + source.addEventListener(sseEventName, listener); + }); + } + + /** + * ensureEventSourceOnElement creates a new EventSource connection on the provided element. + * If a usable EventSource already exists, then it is returned. If not, then a new EventSource + * is created and stored in the element's internalData. + * @param {HTMLElement} elt + * @param {number} retryCount + * @returns {EventSource | null} + */ + function ensureEventSourceOnElement(elt, retryCount) { + + if (elt == null) { + return null; + } + + // handle extension source creation attribute + queryAttributeOnThisOrChildren(elt, "sse-connect").forEach(function(child) { + var sseURL = api.getAttributeValue(child, "sse-connect"); + if (sseURL == null) { + return; + } + + ensureEventSource(child, sseURL, retryCount); + }); + + // handle legacy sse, remove for HTMX2 + queryAttributeOnThisOrChildren(elt, "hx-sse").forEach(function(child) { + var sseURL = getLegacySSEURL(child); + if (sseURL == null) { + return; + } + + ensureEventSource(child, sseURL, retryCount); + }); + + registerSSE(elt); + } + + function ensureEventSource(elt, url, retryCount) { + var internalData = api.getInternalData(elt); + var existingSource = internalData.sseEventSource; + if (existingSource && existingSource.readyState !== EventSource.CLOSED) { + return; + } + if (existingSource) { + existingSource.close(); + internalData.sseEventSource = null; + } + + var source = htmx.createEventSource(url); + + source.onerror = function(err) { + + // Log an error event + api.triggerErrorEvent(elt, "htmx:sseError", { error: err, source: source }); + + // If parent no longer exists in the document, then clean up this EventSource + if (maybeCloseSSESource(elt)) { + return; + } + + // Otherwise, try to reconnect the EventSource + if (source.readyState === EventSource.CLOSED) { + retryCount = retryCount || 0; + var timeout = Math.random() * Math.pow(2, retryCount) * 500; + window.setTimeout(function() { + ensureEventSourceOnElement(elt, Math.min(7, retryCount + 1)); + }, timeout); + } + }; + + source.onopen = function(evt) { + api.triggerEvent(elt, "htmx:sseOpen", { source: source }); + } + + api.getInternalData(elt).sseEventSource = source; + } + + /** + * maybeCloseSSESource confirms that the parent element still exists. + * If not, then any associated SSE source is closed and the function returns true. + * + * @param {HTMLElement} elt + * @returns boolean + */ + function maybeCloseSSESource(elt) { + if (!api.bodyContains(elt)) { + var source = api.getInternalData(elt).sseEventSource; + if (source != undefined) { + source.close(); + // source = null + return true; + } + } + return false; + } + + /** + * queryAttributeOnThisOrChildren returns all nodes that contain the requested attributeName, INCLUDING THE PROVIDED ROOT ELEMENT. + * + * @param {HTMLElement} elt + * @param {string} attributeName + */ + function queryAttributeOnThisOrChildren(elt, attributeName) { + + var result = []; + + // If the parent element also contains the requested attribute, then add it to the results too. + if (api.hasAttribute(elt, attributeName)) { + result.push(elt); + } + + // Search all child nodes that match the requested attribute + elt.querySelectorAll("[" + attributeName + "], [data-" + attributeName + "]").forEach(function(node) { + result.push(node); + }); + + return result; + } + + /** + * @param {HTMLElement} elt + * @param {string} content + */ + function swap(elt, content) { + + api.withExtensions(elt, function(extension) { + content = extension.transformResponse(content, null, elt); + }); + + var swapSpec = api.getSwapSpecification(elt); + var target = api.getTarget(elt); + var settleInfo = api.makeSettleInfo(elt); + + api.selectAndSwap(swapSpec.swapStyle, target, elt, content, settleInfo); + + settleInfo.elts.forEach(function(elt) { + if (elt.classList) { + elt.classList.add(htmx.config.settlingClass); + } + api.triggerEvent(elt, 'htmx:beforeSettle'); + }); + + // Handle settle tasks (with delay if requested) + if (swapSpec.settleDelay > 0) { + setTimeout(doSettle(settleInfo), swapSpec.settleDelay); + } else { + doSettle(settleInfo)(); + } + } + + /** + * doSettle mirrors much of the functionality in htmx that + * settles elements after their content has been swapped. + * TODO: this should be published by htmx, and not duplicated here + * @param {import("../htmx").HtmxSettleInfo} settleInfo + * @returns () => void + */ + function doSettle(settleInfo) { + + return function() { + settleInfo.tasks.forEach(function(task) { + task.call(); + }); + + settleInfo.elts.forEach(function(elt) { + if (elt.classList) { + elt.classList.remove(htmx.config.settlingClass); + } + api.triggerEvent(elt, 'htmx:afterSettle'); + }); + } + } + + function hasEventSource(node) { + return api.getInternalData(node).sseEventSource != null; + } + +})(); diff --git a/fournos-ui/app/static/htmx.min.js b/fournos-ui/app/static/htmx.min.js new file mode 100644 index 0000000..de5f0f1 --- /dev/null +++ b/fournos-ui/app/static/htmx.min.js @@ -0,0 +1 @@ +(function(e,t){if(typeof define==="function"&&define.amd){define([],t)}else if(typeof module==="object"&&module.exports){module.exports=t()}else{e.htmx=e.htmx||t()}})(typeof self!=="undefined"?self:this,function(){return function(){"use strict";var Q={onLoad:F,process:zt,on:de,off:ge,trigger:ce,ajax:Nr,find:C,findAll:f,closest:v,values:function(e,t){var r=dr(e,t||"post");return r.values},remove:_,addClass:z,removeClass:n,toggleClass:$,takeClass:W,defineExtension:Ur,removeExtension:Br,logAll:V,logNone:j,logger:null,config:{historyEnabled:true,historyCacheSize:10,refreshOnHistoryMiss:false,defaultSwapStyle:"innerHTML",defaultSwapDelay:0,defaultSettleDelay:20,includeIndicatorStyles:true,indicatorClass:"htmx-indicator",requestClass:"htmx-request",addedClass:"htmx-added",settlingClass:"htmx-settling",swappingClass:"htmx-swapping",allowEval:true,allowScriptTags:true,inlineScriptNonce:"",attributesToSettle:["class","style","width","height"],withCredentials:false,timeout:0,wsReconnectDelay:"full-jitter",wsBinaryType:"blob",disableSelector:"[hx-disable], [data-hx-disable]",useTemplateFragments:false,scrollBehavior:"smooth",defaultFocusScroll:false,getCacheBusterParam:false,globalViewTransitions:false,methodsThatUseUrlParams:["get"],selfRequestsOnly:false,ignoreTitle:false,scrollIntoViewOnBoost:true,triggerSpecsCache:null},parseInterval:d,_:t,createEventSource:function(e){return new EventSource(e,{withCredentials:true})},createWebSocket:function(e){var t=new WebSocket(e,[]);t.binaryType=Q.config.wsBinaryType;return t},version:"1.9.12"};var r={addTriggerHandler:Lt,bodyContains:se,canAccessLocalStorage:U,findThisElement:xe,filterValues:yr,hasAttribute:o,getAttributeValue:te,getClosestAttributeValue:ne,getClosestMatch:c,getExpressionVars:Hr,getHeaders:xr,getInputValues:dr,getInternalData:ae,getSwapSpecification:wr,getTriggerSpecs:it,getTarget:ye,makeFragment:l,mergeObjects:le,makeSettleInfo:T,oobSwap:Ee,querySelectorExt:ue,selectAndSwap:je,settleImmediately:nr,shouldCancel:ut,triggerEvent:ce,triggerErrorEvent:fe,withExtensions:R};var w=["get","post","put","delete","patch"];var i=w.map(function(e){return"[hx-"+e+"], [data-hx-"+e+"]"}).join(", ");var S=e("head"),q=e("title"),H=e("svg",true);function e(e,t){return new RegExp("<"+e+"(\\s[^>]*>|>)([\\s\\S]*?)<\\/"+e+">",!!t?"gim":"im")}function d(e){if(e==undefined){return undefined}let t=NaN;if(e.slice(-2)=="ms"){t=parseFloat(e.slice(0,-2))}else if(e.slice(-1)=="s"){t=parseFloat(e.slice(0,-1))*1e3}else if(e.slice(-1)=="m"){t=parseFloat(e.slice(0,-1))*1e3*60}else{t=parseFloat(e)}return isNaN(t)?undefined:t}function ee(e,t){return e.getAttribute&&e.getAttribute(t)}function o(e,t){return e.hasAttribute&&(e.hasAttribute(t)||e.hasAttribute("data-"+t))}function te(e,t){return ee(e,t)||ee(e,"data-"+t)}function u(e){return e.parentElement}function re(){return document}function c(e,t){while(e&&!t(e)){e=u(e)}return e?e:null}function L(e,t,r){var n=te(t,r);var i=te(t,"hx-disinherit");if(e!==t&&i&&(i==="*"||i.split(" ").indexOf(r)>=0)){return"unset"}else{return n}}function ne(t,r){var n=null;c(t,function(e){return n=L(t,e,r)});if(n!=="unset"){return n}}function h(e,t){var r=e.matches||e.matchesSelector||e.msMatchesSelector||e.mozMatchesSelector||e.webkitMatchesSelector||e.oMatchesSelector;return r&&r.call(e,t)}function A(e){var t=/<([a-z][^\/\0>\x20\t\r\n\f]*)/i;var r=t.exec(e);if(r){return r[1].toLowerCase()}else{return""}}function s(e,t){var r=new DOMParser;var n=r.parseFromString(e,"text/html");var i=n.body;while(t>0){t--;i=i.firstChild}if(i==null){i=re().createDocumentFragment()}return i}function N(e){return/
"+n+"",0);var a=i.querySelector("template").content;if(Q.config.allowScriptTags){oe(a.querySelectorAll("script"),function(e){if(Q.config.inlineScriptNonce){e.nonce=Q.config.inlineScriptNonce}e.htmxExecuted=navigator.userAgent.indexOf("Firefox")===-1})}else{oe(a.querySelectorAll("script"),function(e){_(e)})}return a}switch(r){case"thead":case"tbody":case"tfoot":case"colgroup":case"caption":return s("{{ job.spec.pipeline }}| Name | +Status | +Ready | +Restarts | +Age | +Logs | +
|---|---|---|---|---|---|
| {{ pod.name }} | ++ + {% if pod.phase == "Running" %}{% endif %} + {{ pod.phase }} + + | +{{ "Yes" if pod.ready else "No" }} | +{{ pod.restarts }} | +{{ pod.age_minutes }}m | ++ {% if phase in ("Running", "Pending", "Admitted") %} + View logs + {% else %} + - + {% endif %} + | +
{{ info.args | join(", ") or "default" }}{{ job.spec.pipeline }}{{ job.spec.cluster }}{{ info.args | join(", ") or "none" }}{{ k }}: {{ v }}{% if not loop.last %}{{ s }}{% if not loop.last %}, {% endif %}
+ {% endfor %}
+ | Status | +Job | +Project | +Preset | +Owner | +Cluster | +Progress | +Age | +PR | ++ |
|---|
No live FournosJobs found.
+ Submit a new job +| Status | +Job | +Project | +Preset | +Owner | +Cluster | +Trigger | +Duration | +Date | +MLflow | ++ |
|---|---|---|---|---|---|---|---|---|---|---|
| + {{ job.phase }} + | ++ {{ job.name }} + | +{{ job.project }} | +{{ job.preset or "default" }} |
+ {{ job.owner }} | +{{ job.cluster }} | ++ {% if job.trigger_type == "scheduled" %} + scheduled + {% elif job.triggered_by_schedule %} + manual + {% else %} + direct + {% endif %} + {% if job.triggered_by_schedule %} + {{ job.triggered_by_schedule }} + {% endif %} + | +{{ format_duration(job.duration_seconds) }} | +{{ format_age(job.created_at) }} | ++ {% if job.mlflow_url %} + View + {% else %} + - + {% endif %} + | +
+
+
+
+
+ |
+
| + No archived jobs found. + | +||||||||||
{{ schedule_name }}| Status | +Job Name | +Preset | +Trigger | +Duration | +MLflow | +Created | +
|---|---|---|---|---|---|---|
| {{ run.status }} | +{{ run.name }} |
+ {{ run.preset or "default" }} |
+ + {% if run.trigger_type == "scheduled" %} + scheduled + {% else %} + manual + {% endif %} + | ++ {% if run.duration_seconds %} + {{ format_duration(run.duration_seconds) }} + {% else %} + -- + {% endif %} + | ++ {% if run.mlflow_url %} + View in MLflow + {% else %} + -- + {% endif %} + | ++ {% if run.created_at %} + {{ format_age(run.created_at) }} ago + {% else %} + -- + {% endif %} + | +
No runs found for this schedule yet.
+Manage CronJob-based scheduled FournosJob submissions.
+| Status | +Name | +Project | +Preset | +Schedule | +Cluster | +Resolver | +Owner | +Last Run | +Actions | +
|---|---|---|---|---|---|---|---|---|---|
| + {% if cj.suspend %} + Paused + {% else %} + Active + {% endif %} + | +{{ cj.name }} |
+ {{ cj.project }} | +{{ cj.preset or "default" }} |
+ {{ cj.schedule }} |
+ {{ cj.cluster }} | +
+ {% if cj.has_resolver %}
+
+ View
+
+
+ {% else %}
+ --
+ {% endif %}
+ |
+ {{ cj.owner }} | ++ {% if cj.last_schedule %} + {{ format_age(cj.last_schedule) }} ago + {% else %} + Never + {% endif %} + | ++ + | +
No scheduled runs configured.
+Create a new FournosJob on the cluster.
+