Skip to content

Commit 4dd9810

Browse files
ntkatholentkathole
committed
feat: Added run history
Signed-off-by: ntkathole <nikhilkathole2683@gmail.com>
1 parent d752eb0 commit 4dd9810

6 files changed

Lines changed: 509 additions & 1 deletion

File tree

‎docs/reference/openlineage.md‎

Lines changed: 57 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -311,6 +311,12 @@ When the consumer is enabled, the following endpoints are available on the Feast
311311

312312
Both endpoints require the `X-API-Key` header (or `Authorization: Bearer <key>`) if `consumer.api_key` is configured.
313313

314+
#### Admin Endpoints
315+
316+
| Endpoint | Method | Description |
317+
|----------|--------|-------------|
318+
| `/lineage/openlineage/reset` | `DELETE` | Purge all OpenLineage data. Accepts optional `?namespace=X` to delete only a specific namespace. Requires API key. |
319+
314320
#### OpenLineage Query Endpoints (UI-facing)
315321

316322
| Endpoint | Method | Description |
@@ -320,6 +326,8 @@ Both endpoints require the `X-API-Key` header (or `Authorization: Bearer <key>`)
320326
| `/lineage/openlineage/events` | `GET` | Browse stored events with filtering |
321327
| `/lineage/openlineage/jobs` | `GET` | List all known OpenLineage jobs |
322328
| `/lineage/openlineage/datasets` | `GET` | List all known OpenLineage datasets |
329+
| `/lineage/openlineage/runs` | `GET` | List runs with optional `?job_namespace=X&job_name=Y` filtering |
330+
| `/lineage/openlineage/runs/{run_id}` | `GET` | Single run detail with input/output datasets |
323331

324332
#### Registry Query Endpoints
325333

@@ -381,7 +389,7 @@ When the consumer is enabled, the lineage page in the Feast UI shows two tabs:
381389

382390
**Lineage tab**
383391

384-
- **OpenLineage Graph** (default) — shows lineage from all OpenLineage producers with cross-producer connectivity. Nodes are color-coded by producer (colors generated dynamically). The graph supports filtering by type, producer, and object name. Clicking a node opens a **detail panel** showing description, schema, tags, features, entities, data quality metrics, data source info, and other facets.
392+
- **OpenLineage Graph** (default) — shows lineage from all OpenLineage producers with cross-producer connectivity. Nodes are color-coded by producer (colors generated dynamically). The graph supports filtering by type, producer, and object name. Clicking a node opens a **detail panel** showing description, schema, tags, features, entities, data quality metrics, data source info, other facets, and **run history** (for job nodes — see [Per-Run Lineage](#per-run-lineage-run-history)).
385393
- **Feast Only Lineage** (checkbox) — switches to the original Feast registry view (DataSource → FeatureView → FeatureService) powered entirely by the Feast registry.
386394

387395
**Events tab**
@@ -405,6 +413,54 @@ The OpenLineage consumer integrates with Feast's existing RBAC:
405413
- **Write access** (producers sending events): Authenticated via API key in the `X-API-Key` header
406414
- **Read access** (UI viewing lineage): Namespace-based filtering maps OpenLineage namespaces to Feast projects. Users see only lineage data for namespaces they have access to via the `namespace_mapping` configuration
407415

416+
### Lineage Cleanup / Reset
417+
418+
Over time the OpenLineage store accumulates historical data. Two mechanisms are provided for cleanup:
419+
420+
#### Admin Reset Endpoint
421+
422+
Use the `DELETE /lineage/openlineage/reset` endpoint to purge lineage data. The endpoint requires the same API key used for event ingestion.
423+
424+
```bash
425+
# Purge ALL OpenLineage data
426+
curl -X DELETE -H "X-API-Key: your-key" \
427+
http://localhost:8080/api/v1/lineage/openlineage/reset
428+
429+
# Purge only a specific namespace
430+
curl -X DELETE -H "X-API-Key: your-key" \
431+
"http://localhost:8080/api/v1/lineage/openlineage/reset?namespace=airflow://prod-cluster"
432+
```
433+
434+
A full purge deletes data from all seven `openlineage_*` tables. A namespace-scoped purge deletes jobs, datasets, runs, events, edges, and symlinks associated with that namespace, leaving other namespaces intact.
435+
436+
#### Feast Teardown Hook
437+
438+
When you run `feast teardown`, Feast automatically cleans up OpenLineage data for the project's namespace (if the consumer is configured). This ensures that tearing down a Feast project doesn't leave orphaned lineage data behind.
439+
440+
```bash
441+
# Tears down the Feast project AND its OpenLineage lineage
442+
feast teardown
443+
```
444+
445+
### Per-Run Lineage (Run History)
446+
447+
The consumer tracks individual pipeline runs in the `openlineage_runs` table. When you click on a **job node** in the OpenLineage Graph, the detail panel shows a **Run History** section with:
448+
449+
- A table of past runs: truncated run ID, status badge (COMPLETE, FAIL, RUNNING, ABORT), start time, and duration
450+
- Click any run to expand its **inputs and outputs** — the specific datasets that run consumed and produced
451+
452+
#### Run History API
453+
454+
```bash
455+
# List runs for a specific job
456+
curl "http://localhost:8080/api/v1/lineage/openlineage/runs?job_namespace=spark://emr-cluster&job_name=feature_engineering"
457+
458+
# Get a single run with its I/O datasets
459+
curl "http://localhost:8080/api/v1/lineage/openlineage/runs/{run_id}"
460+
```
461+
462+
The run detail response includes `inputs` and `outputs` arrays, each containing the dataset namespace, name, and any I/O facets recorded by the producer.
463+
408464
### Database Schema
409465

410466
The consumer creates the following tables (automatically on first startup):

‎sdk/python/feast/feature_store.py‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1675,6 +1675,28 @@ def teardown(self):
16751675

16761676
self._get_provider().teardown_infra(self.project, tables, entities) # type: ignore[arg-type]
16771677
self.registry.teardown()
1678+
self._teardown_openlineage()
1679+
1680+
def _teardown_openlineage(self):
1681+
"""Clean up OpenLineage data for this project's namespace during teardown."""
1682+
try:
1683+
if (
1684+
hasattr(self.config, "openlineage")
1685+
and self.config.openlineage is not None
1686+
and self.config.openlineage.enabled
1687+
):
1688+
ol_config = self.config.openlineage.to_openlineage_config()
1689+
consumer_cfg = getattr(ol_config, "consumer", None)
1690+
if consumer_cfg and getattr(consumer_cfg, "enabled", False):
1691+
conn_str = getattr(consumer_cfg, "connection_string", None)
1692+
if conn_str:
1693+
from feast.openlineage.store import OpenLineageStore
1694+
1695+
ol_store = OpenLineageStore(connection_string=conn_str)
1696+
namespace = f"{self.project}/{self.project}"
1697+
ol_store.purge_namespace(namespace)
1698+
except Exception as e:
1699+
warnings.warn(f"Failed to clean up OpenLineage data during teardown: {e}")
16781700

16791701
def get_historical_features(
16801702
self,

‎sdk/python/feast/openlineage/consumer.py‎

Lines changed: 51 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -166,6 +166,31 @@ async def receive_lineage_batch(
166166
},
167167
)
168168

169+
# ── Admin endpoints ──
170+
171+
@router.delete("/lineage/openlineage/reset")
172+
async def reset_lineage(
173+
namespace: Optional[str] = Query(None),
174+
x_api_key: Optional[str] = Header(None, alias="X-API-Key"),
175+
authorization: Optional[str] = Header(None),
176+
):
177+
"""
178+
Purge OpenLineage data. Requires API key.
179+
180+
If ?namespace=X is provided, only that namespace's data is deleted.
181+
Otherwise, all OpenLineage data is purged.
182+
"""
183+
api_key = getattr(config, "consumer_api_key", None)
184+
if not _verify_api_key(api_key, x_api_key, authorization):
185+
raise HTTPException(status_code=401, detail="Invalid API key")
186+
187+
if namespace:
188+
store.purge_namespace(namespace)
189+
return {"status": "success", "message": f"Purged namespace: {namespace}"}
190+
else:
191+
store.purge_all()
192+
return {"status": "success", "message": "Purged all OpenLineage data"}
193+
169194
# ── Query endpoints ──
170195

171196
@router.get("/lineage/openlineage/events")
@@ -283,6 +308,32 @@ def get_full_lineage_graph():
283308

284309
return {"nodes": nodes, "edges": edges, "symlinks": symlinks}
285310

311+
# ── Run history endpoints ──
312+
313+
@router.get("/lineage/openlineage/runs")
314+
def list_runs(
315+
job_namespace: Optional[str] = Query(None),
316+
job_name: Optional[str] = Query(None),
317+
limit: int = Query(50, ge=1, le=500),
318+
offset: int = Query(0, ge=0),
319+
):
320+
"""List runs, optionally filtered by job namespace and name."""
321+
runs = store.get_runs(
322+
job_namespace=job_namespace,
323+
job_name=job_name,
324+
limit=limit,
325+
offset=offset,
326+
)
327+
return {"runs": runs, "total": len(runs)}
328+
329+
@router.get("/lineage/openlineage/runs/{run_id}")
330+
def get_run_detail(run_id: str):
331+
"""Get a single run with its input and output datasets."""
332+
run = store.get_run_detail(run_id)
333+
if not run:
334+
raise HTTPException(status_code=404, detail=f"Run {run_id} not found")
335+
return run
336+
286337
def _get_namespace_filter(ns_callable) -> Optional[List[str]]:
287338
if ns_callable:
288339
try:

‎sdk/python/feast/openlineage/store.py‎

Lines changed: 120 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -652,6 +652,117 @@ def get_all_symlinks(self) -> List[Dict[str, Any]]:
652652
rows = conn.execute(select(tbl)).fetchall()
653653
return [dict(r._mapping) for r in rows]
654654

655+
# ── Cleanup methods ──
656+
657+
def purge_all(self):
658+
"""Delete all data from all OpenLineage tables."""
659+
table_order = [
660+
"run_io",
661+
"runs",
662+
"lineage_edges",
663+
"dataset_symlinks",
664+
"events",
665+
"datasets",
666+
"jobs",
667+
]
668+
with self._engine.begin() as conn:
669+
for tbl_name in table_order:
670+
conn.execute(OL_TABLES[tbl_name].delete())
671+
logger.info("Purged all OpenLineage data")
672+
673+
def purge_namespace(self, namespace: str):
674+
"""Delete all data associated with a specific namespace."""
675+
with self._engine.begin() as conn:
676+
tbl_runs = OL_TABLES["runs"]
677+
run_ids_q = select(tbl_runs.c.run_id).where(
678+
tbl_runs.c.job_namespace == namespace
679+
)
680+
run_ids = [r[0] for r in conn.execute(run_ids_q).fetchall()]
681+
if run_ids:
682+
tbl_rio = OL_TABLES["run_io"]
683+
conn.execute(tbl_rio.delete().where(tbl_rio.c.run_id.in_(run_ids)))
684+
conn.execute(tbl_runs.delete().where(tbl_runs.c.run_id.in_(run_ids)))
685+
686+
tbl_ev = OL_TABLES["events"]
687+
conn.execute(tbl_ev.delete().where(tbl_ev.c.job_namespace == namespace))
688+
689+
tbl_edges = OL_TABLES["lineage_edges"]
690+
conn.execute(
691+
tbl_edges.delete().where(
692+
(tbl_edges.c.source_namespace == namespace)
693+
| (tbl_edges.c.target_namespace == namespace)
694+
)
695+
)
696+
697+
tbl_sym = OL_TABLES["dataset_symlinks"]
698+
conn.execute(
699+
tbl_sym.delete().where(
700+
(tbl_sym.c.dataset_namespace == namespace)
701+
| (tbl_sym.c.linked_namespace == namespace)
702+
)
703+
)
704+
705+
tbl_ds = OL_TABLES["datasets"]
706+
conn.execute(tbl_ds.delete().where(tbl_ds.c.dataset_namespace == namespace))
707+
708+
tbl_jobs = OL_TABLES["jobs"]
709+
conn.execute(tbl_jobs.delete().where(tbl_jobs.c.job_namespace == namespace))
710+
711+
logger.info(f"Purged OpenLineage data for namespace: {namespace}")
712+
713+
# ── Run query methods ──
714+
715+
def get_runs(
716+
self,
717+
job_namespace: Optional[str] = None,
718+
job_name: Optional[str] = None,
719+
limit: int = 50,
720+
offset: int = 0,
721+
) -> List[Dict[str, Any]]:
722+
"""Get runs, optionally filtered by job."""
723+
tbl = OL_TABLES["runs"]
724+
query = (
725+
select(tbl).order_by(tbl.c.updated_at.desc()).limit(limit).offset(offset)
726+
)
727+
if job_namespace:
728+
query = query.where(tbl.c.job_namespace == job_namespace)
729+
if job_name:
730+
query = query.where(tbl.c.job_name == job_name)
731+
with self._engine.connect() as conn:
732+
rows = conn.execute(query).fetchall()
733+
return [dict(row._mapping) for row in rows]
734+
735+
def get_run_detail(self, run_id: str) -> Optional[Dict[str, Any]]:
736+
"""Get a single run with its I/O datasets."""
737+
tbl_runs = OL_TABLES["runs"]
738+
tbl_rio = OL_TABLES["run_io"]
739+
with self._engine.connect() as conn:
740+
run_row = conn.execute(
741+
select(tbl_runs).where(tbl_runs.c.run_id == run_id)
742+
).first()
743+
if not run_row:
744+
return None
745+
run = dict(run_row._mapping)
746+
747+
io_rows = conn.execute(
748+
select(tbl_rio).where(tbl_rio.c.run_id == run_id)
749+
).fetchall()
750+
run["inputs"] = []
751+
run["outputs"] = []
752+
for io_row in io_rows:
753+
io = dict(io_row._mapping)
754+
entry = {
755+
"namespace": io["dataset_namespace"],
756+
"name": io["dataset_name"],
757+
"facets": _safe_parse_json(io.get("facets_json")),
758+
}
759+
if io["io_type"] == "INPUT":
760+
run["inputs"].append(entry)
761+
else:
762+
run["outputs"].append(entry)
763+
run["facets"] = _safe_parse_json(run.pop("facets_json", None))
764+
return run
765+
655766
def get_all_lineage_edges(
656767
self, namespaces: Optional[List[str]] = None
657768
) -> List[Dict[str, Any]]:
@@ -667,6 +778,15 @@ def get_all_lineage_edges(
667778
return [dict(row._mapping) for row in rows]
668779

669780

781+
def _safe_parse_json(val: Optional[str]) -> Optional[Any]:
782+
if not val:
783+
return None
784+
try:
785+
return json.loads(val)
786+
except (json.JSONDecodeError, TypeError):
787+
return None
788+
789+
670790
def _parse_timestamp(ts_str: str) -> int:
671791
if not ts_str:
672792
return int(time.time() * 1000)

0 commit comments

Comments
 (0)