Skip to content

Commit 3ad7895

Browse files
ntkatholentkathole
committed
fix: Fixed OpenLineage actions via apis and apply
Signed-off-by: ntkathole <nikhilkathole2683@gmail.com>
1 parent ff298df commit 3ad7895

19 files changed

Lines changed: 1418 additions & 360 deletions

File tree

‎docs/reference/openlineage.md‎

Lines changed: 467 additions & 196 deletions
Large diffs are not rendered by default.

‎sdk/python/feast/api/registry/rest/__init__.py‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -159,6 +159,14 @@ def _build_allowed_namespaces_fn(fs):
159159
Uses ``permitted_resources`` with ``DESCRIBE`` on all known projects.
160160
The projects the current user may describe become the allowed OL
161161
namespaces (mapped through ``resolve_namespace``).
162+
163+
External producer namespaces (e.g. ``spark://ml-team``,
164+
``airflow://prod-cluster``) are included when
165+
``consumer.namespace_mapping`` maps them to a Feast project the
166+
user is allowed to DESCRIBE. This is the only purpose of
167+
``namespace_mapping`` — it is a **read-side RBAC bridge**, not a
168+
routing or rewrite mechanism. Ingest stores events as-is;
169+
producers own their namespace.
162170
"""
163171

164172
def _get_allowed():
@@ -183,9 +191,20 @@ def _get_allowed():
183191
)
184192

185193
ol_ns_config = getattr(ol_config, "namespace", "feast")
194+
allowed_project_names = {p.name for p in allowed_projects}
195+
186196
namespaces = set()
187197
for p in allowed_projects:
188198
namespaces.add(resolve_namespace(ol_ns_config, p.name))
199+
200+
# Include external namespaces whose namespace_mapping
201+
# target is a project the user can DESCRIBE.
202+
consumer_cfg = getattr(ol_config, "consumer", None)
203+
ns_map = getattr(consumer_cfg, "namespace_mapping", None) or {}
204+
for ext_ns, mapped_project in ns_map.items():
205+
if mapped_project in allowed_project_names:
206+
namespaces.add(ext_ns)
207+
189208
return list(namespaces) if namespaces else None
190209
except Exception:
191210
return None

‎sdk/python/feast/feature_store.py‎

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1907,9 +1907,18 @@ def _mlflow_log_apply(
19071907
_logger.debug("MLflow apply logging failed: %s", e)
19081908

19091909
def _emit_openlineage_apply(self, objects: List[Any]):
1910-
"""Emit OpenLineage events for applied objects."""
1910+
"""Emit OpenLineage events for applied objects.
1911+
1912+
Skips when using a remote registry — the RegistryServer already
1913+
emits OL events in its Apply* handlers, so emitting here would
1914+
double-count every object.
1915+
"""
19111916
if self.openlineage_emitter is None:
19121917
return
1918+
from feast.infra.registry.remote import RemoteRegistry
1919+
1920+
if isinstance(self._registry, RemoteRegistry):
1921+
return
19131922
try:
19141923
self.openlineage_emitter.emit_apply(objects, self.project)
19151924
except Exception as e:

‎sdk/python/feast/lineage/registry_lineage.py‎

Lines changed: 29 additions & 50 deletions
Original file line numberDiff line numberDiff line change
@@ -314,59 +314,38 @@ def _parse_direct_relationships(self, registry: Registry) -> List[EntityRelation
314314
source_items = [(k, v) for k, v in enumerate(odfv.spec.sources)]
315315

316316
for source_name, source in source_items:
317-
if (
318-
hasattr(source, "request_data_source")
319-
and source.request_data_source
320-
):
321-
if hasattr(source.request_data_source, "name"):
322-
relationships.append(
323-
EntityRelation(
324-
source=EntityReference(
325-
FeastObjectType.DATA_SOURCE,
326-
source.request_data_source.name,
327-
),
328-
target=EntityReference(
329-
FeastObjectType.FEATURE_VIEW, odfv.spec.name
330-
),
331-
)
317+
has_req = hasattr(source, "HasField") and source.HasField(
318+
"request_data_source"
319+
)
320+
has_fvp = hasattr(source, "HasField") and source.HasField(
321+
"feature_view_projection"
322+
)
323+
324+
if has_req and source.request_data_source.name:
325+
relationships.append(
326+
EntityRelation(
327+
source=EntityReference(
328+
FeastObjectType.DATA_SOURCE,
329+
source.request_data_source.name,
330+
),
331+
target=EntityReference(
332+
FeastObjectType.FEATURE_VIEW, odfv.spec.name
333+
),
332334
)
333-
elif (
334-
hasattr(source, "feature_view_projection")
335-
and source.feature_view_projection
336-
):
337-
# Find the source feature view's batch source
338-
if hasattr(source.feature_view_projection, "feature_view_name"):
339-
source_fv = next(
340-
(
341-
fv
342-
for fv in registry.feature_views
343-
if hasattr(fv, "spec")
344-
and fv.spec
345-
and hasattr(fv.spec, "name")
346-
and fv.spec.name
347-
== source.feature_view_projection.feature_view_name
335+
)
336+
elif has_fvp and source.feature_view_projection.feature_view_name:
337+
relationships.append(
338+
EntityRelation(
339+
source=EntityReference(
340+
FeastObjectType.FEATURE_VIEW,
341+
source.feature_view_projection.feature_view_name,
342+
),
343+
target=EntityReference(
344+
FeastObjectType.FEATURE_VIEW,
345+
odfv.spec.name,
348346
),
349-
None,
350347
)
351-
if (
352-
source_fv
353-
and hasattr(source_fv, "spec")
354-
and source_fv.spec
355-
and hasattr(source_fv.spec, "batch_source")
356-
and source_fv.spec.batch_source
357-
and hasattr(source_fv.spec.batch_source, "name")
358-
):
359-
relationships.append(
360-
EntityRelation(
361-
source=EntityReference(
362-
FeastObjectType.DATA_SOURCE,
363-
source_fv.spec.batch_source.name,
364-
),
365-
target=EntityReference(
366-
FeastObjectType.FEATURE_VIEW, odfv.spec.name
367-
),
368-
)
369-
)
348+
)
370349

371350
# Stream FeatureView relationships
372351
for sfv in registry.stream_feature_views:

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

Lines changed: 19 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616
Configuration classes for Feast OpenLineage integration.
1717
"""
1818

19+
import json
1920
import os
2021
from dataclasses import dataclass, field
2122
from typing import Any, Dict, Optional
@@ -31,7 +32,12 @@ class OpenLineageConsumerConfig:
3132
store_type: Storage backend type ('sql' uses the SQL registry DB)
3233
connection_string: Optional separate DB connection string
3334
api_key: API key for authenticating producers sending events
34-
namespace_mapping: Map of OL namespace -> Feast project for RBAC scoping
35+
namespace_mapping: Read-side RBAC bridge mapping external OpenLineage
36+
namespaces to Feast project names. When a user can DESCRIBE a Feast
37+
project, they also see lineage from any external namespace mapped to
38+
that project. Also used during ingest to resolve incoming datasets to
39+
Feast registry objects. Example:
40+
{"spark://ml-team": "ml_team", "airflow://prod-cluster": "ml_team"}
3541
"""
3642

3743
enabled: bool = False
@@ -139,16 +145,28 @@ def from_env(cls) -> "OpenLineageConfig":
139145
FEAST_OPENLINEAGE_API_KEY: API key for authentication
140146
FEAST_OPENLINEAGE_NAMESPACE: Default namespace (default: feast)
141147
FEAST_OPENLINEAGE_PRODUCER: Producer identifier
148+
FEAST_OPENLINEAGE_CONSUMER_NAMESPACE_MAPPING: JSON object mapping external
149+
OL namespaces to Feast project names for RBAC scoping.
150+
Example: '{"spark://ml-team": "ml_team", "airflow://prod-cluster": "prod"}'
142151
143152
Returns:
144153
OpenLineageConfig instance
145154
"""
155+
ns_mapping_raw = os.getenv("FEAST_OPENLINEAGE_CONSUMER_NAMESPACE_MAPPING", "")
156+
ns_mapping: Dict[str, str] = {}
157+
if ns_mapping_raw:
158+
try:
159+
ns_mapping = json.loads(ns_mapping_raw)
160+
except json.JSONDecodeError:
161+
pass
162+
146163
consumer = OpenLineageConsumerConfig(
147164
enabled=os.getenv("FEAST_OPENLINEAGE_CONSUMER_ENABLED", "false").lower()
148165
== "true",
149166
store_type=os.getenv("FEAST_OPENLINEAGE_CONSUMER_STORE_TYPE", "sql"),
150167
connection_string=os.getenv("FEAST_OPENLINEAGE_CONSUMER_CONNECTION_STRING"),
151168
api_key=os.getenv("FEAST_OPENLINEAGE_CONSUMER_API_KEY"),
169+
namespace_mapping=ns_mapping,
152170
)
153171

154172
return cls(

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

Lines changed: 20 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -86,7 +86,7 @@ def get_consumer_router(
8686

8787
# ── Producer-facing: receive events ──
8888

89-
@router.post("/v1/lineage")
89+
@router.post("/lineage")
9090
async def receive_lineage_event(
9191
request: Request,
9292
x_api_key: Optional[str] = Header(None, alias="X-API-Key"),
@@ -97,6 +97,13 @@ async def receive_lineage_event(
9797
9898
Compatible with the standard OpenLineage API endpoint.
9999
Accepts RunEvent, DatasetEvent, or JobEvent.
100+
101+
The router defines POST /lineage; the full path depends on
102+
the server mount:
103+
- UI server (mounted at /api/v1): POST /api/v1/lineage
104+
- REST server (root_path=/api/v1): POST /lineage
105+
When behind a reverse proxy, the proxy strips /api/v1,
106+
so external clients use POST /api/v1/lineage in both cases.
100107
"""
101108
api_key = getattr(config, "consumer_api_key", None)
102109
if not _verify_api_key(api_key, x_api_key, authorization):
@@ -130,7 +137,7 @@ async def receive_lineage_event(
130137
logger.error(f"Failed to process event: {e}")
131138
raise HTTPException(status_code=500, detail=str(e))
132139

133-
@router.post("/v1/lineage/batch")
140+
@router.post("/lineage/batch")
134141
async def receive_lineage_batch(
135142
request: Request,
136143
x_api_key: Optional[str] = Header(None, alias="X-API-Key"),
@@ -202,6 +209,8 @@ def list_events(
202209
):
203210
"""List stored OpenLineage events with optional filtering."""
204211
ns_filter = _get_namespace_filter(get_allowed_namespaces)
212+
if namespace and ns_filter is not None and namespace not in ns_filter:
213+
return {"events": [], "total": 0}
205214
events = store.get_events(
206215
namespace=namespace,
207216
job_name=job_name,
@@ -362,11 +371,15 @@ def list_runs(
362371
offset: int = Query(0, ge=0),
363372
):
364373
"""List runs, optionally filtered by job namespace and name."""
374+
ns_filter = _get_namespace_filter(get_allowed_namespaces)
375+
if job_namespace and ns_filter is not None and job_namespace not in ns_filter:
376+
return {"runs": [], "total": 0}
365377
runs = store.get_runs(
366378
job_namespace=job_namespace,
367379
job_name=job_name,
368380
limit=limit,
369381
offset=offset,
382+
namespaces=ns_filter if not job_namespace else None,
370383
)
371384
return {"runs": runs, "total": len(runs)}
372385

@@ -376,6 +389,11 @@ def get_run_detail(run_id: str):
376389
run = store.get_run_detail(run_id)
377390
if not run:
378391
raise HTTPException(status_code=404, detail=f"Run {run_id} not found")
392+
ns_filter = _get_namespace_filter(get_allowed_namespaces)
393+
if ns_filter is not None:
394+
run_ns = run.get("job_namespace")
395+
if run_ns and run_ns not in ns_filter:
396+
raise HTTPException(status_code=404, detail=f"Run {run_id} not found")
379397
return run
380398

381399
def _get_namespace_filter(ns_callable) -> Optional[List[str]]:

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

Lines changed: 22 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -445,15 +445,29 @@ def emit_on_demand_feature_view_lineage(
445445
)
446446

447447
for source_name, req_source in odfv.source_request_sources.items():
448+
req_name = getattr(req_source, "name", source_name)
448449
inputs.append(
449450
InputDataset(
450451
namespace=namespace,
451-
name=f"request_source_{source_name}",
452+
name=req_name,
452453
)
453454
)
454455

455-
# Build output
456-
output_facets = {}
456+
# Build output with feast_featureView facet on the dataset
457+
output_facets: Dict[str, Any] = {
458+
"feast_featureView": FeastFeatureViewFacet(
459+
name=odfv.name,
460+
ttl_seconds=0,
461+
entities=[],
462+
features=[f.name for f in odfv.features] if odfv.features else [],
463+
online_enabled=True,
464+
offline_enabled=True,
465+
mode="ON_DEMAND",
466+
description=odfv.description if odfv.description else "",
467+
owner=odfv.owner if hasattr(odfv, "owner") and odfv.owner else "",
468+
tags=odfv.tags if odfv.tags else {},
469+
),
470+
}
457471
if odfv.features:
458472
output_facets["schema"] = schema_dataset.SchemaDatasetFacet(
459473
fields=[feast_field_to_schema_field(f) for f in odfv.features]
@@ -467,20 +481,12 @@ def emit_on_demand_feature_view_lineage(
467481
)
468482
]
469483

470-
# Build job facets
484+
from feast.openlineage.facets import FeastProjectFacet
485+
from feast.openlineage.identity import FeastJobKind
486+
471487
job_facets = {
472-
"feast_featureView": FeastFeatureViewFacet(
473-
name=odfv.name,
474-
ttl_seconds=0,
475-
entities=[],
476-
features=[f.name for f in odfv.features] if odfv.features else [],
477-
online_enabled=True,
478-
offline_enabled=True,
479-
mode="ON_DEMAND",
480-
description=odfv.description if odfv.description else "",
481-
owner=odfv.owner if hasattr(odfv, "owner") and odfv.owner else "",
482-
tags=odfv.tags if odfv.tags else {},
483-
)
488+
"feast_project": FeastProjectFacet(project_name=project),
489+
**self._job_kind_facets(FeastJobKind.DEFINITION, project),
484490
}
485491

486492
# Emit a RunEvent with COMPLETE state to create lineage connection

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

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -147,6 +147,9 @@ class FeastDataSourceFacet(DatasetFacet):
147147
timestamp_field: Name of the timestamp field
148148
created_timestamp_field: Name of the created timestamp field
149149
field_mapping: Mapping from source fields to feature names
150+
path: File path (for file-based sources)
151+
table: Table name (for database sources)
152+
query: SQL query (for query-based sources)
150153
description: Human-readable description
151154
tags: Key-value tags
152155
"""
@@ -156,6 +159,9 @@ class FeastDataSourceFacet(DatasetFacet):
156159
timestamp_field: Optional[str] = attr.field(default=None)
157160
created_timestamp_field: Optional[str] = attr.field(default=None)
158161
field_mapping: Dict[str, str] = attr.field(factory=dict)
162+
path: Optional[str] = attr.field(default=None)
163+
table: Optional[str] = attr.field(default=None)
164+
query: Optional[str] = attr.field(default=None)
159165
description: str = attr.field(default="")
160166
tags: Dict[str, str] = attr.field(factory=dict)
161167

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

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -139,6 +139,15 @@ def data_source_to_dataset(
139139
field_mapping=data_source.field_mapping
140140
if hasattr(data_source, "field_mapping")
141141
else {},
142+
path=data_source.path
143+
if hasattr(data_source, "path") and data_source.path
144+
else None,
145+
table=data_source.table
146+
if hasattr(data_source, "table") and data_source.table
147+
else None,
148+
query=data_source.query
149+
if hasattr(data_source, "query") and data_source.query
150+
else None,
142151
description=data_source.description
143152
if hasattr(data_source, "description")
144153
else "",

0 commit comments

Comments
 (0)