Skip to content

Commit ba2c05c

Browse files
ntkatholentkathole
committed
feat: Added compute and jobs UI
Signed-off-by: ntkathole <nikhilkathole2683@gmail.com>
1 parent e8f2000 commit ba2c05c

9 files changed

Lines changed: 978 additions & 1 deletion

File tree

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

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33

44
from fastapi import FastAPI
55

6+
from feast.api.registry.rest.compute_engines import get_compute_engine_router
67
from feast.api.registry.rest.data_sources import get_data_source_router
78
from feast.api.registry.rest.entities import get_entity_router
89
from feast.api.registry.rest.feature_services import get_feature_service_router
@@ -41,6 +42,7 @@ def register_all_routes(app: FastAPI, grpc_handler, server=None, store=None):
4142
server.store if server and hasattr(server, "store") else None
4243
)
4344
app.include_router(get_monitoring_router(grpc_handler, store=resolved_store))
45+
app.include_router(get_compute_engine_router(grpc_handler, store=resolved_store))
4446

4547
_register_openlineage_consumer(app, resolved_store)
4648

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

Lines changed: 256 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,256 @@
1+
import logging
2+
from typing import Any, Dict, List, Optional
3+
4+
from fastapi import APIRouter, Depends, Query
5+
6+
from feast.api.registry.rest.rest_utils import (
7+
get_pagination_params,
8+
get_sorting_params,
9+
grpc_call,
10+
)
11+
from feast.protos.feast.registry import RegistryServer_pb2
12+
from feast.repo_config import BATCH_ENGINE_CLASS_FOR_TYPE
13+
14+
logger = logging.getLogger(__name__)
15+
16+
17+
def _extract_engine_info(store) -> Dict[str, Any]:
18+
"""Extract compute engine configuration from the FeatureStore."""
19+
config = getattr(store, "config", None)
20+
if config is None:
21+
return {
22+
"engineType": "local",
23+
"engineClass": "LocalComputeEngine",
24+
"config": {"type": "local"},
25+
}
26+
27+
batch_engine = getattr(config, "batch_engine", None)
28+
if batch_engine is None:
29+
return {
30+
"engineType": "local",
31+
"engineClass": "LocalComputeEngine",
32+
"config": {"type": "local"},
33+
}
34+
35+
if isinstance(batch_engine, dict):
36+
engine_type = batch_engine.get("type", "local")
37+
engine_config = {k: v for k, v in batch_engine.items() if v is not None}
38+
elif hasattr(batch_engine, "type"):
39+
engine_type = str(batch_engine.type)
40+
dump = getattr(batch_engine, "model_dump", None) or getattr(
41+
batch_engine, "dict", None
42+
)
43+
if dump:
44+
engine_config = {k: v for k, v in dump().items() if v is not None}
45+
else:
46+
engine_config = {"type": engine_type}
47+
else:
48+
engine_type = str(batch_engine)
49+
engine_config = {"type": engine_type}
50+
51+
engine_class_path = BATCH_ENGINE_CLASS_FOR_TYPE.get(engine_type, engine_type)
52+
engine_class = (
53+
engine_class_path.rsplit(".", 1)[-1]
54+
if "." in engine_class_path
55+
else engine_class_path
56+
)
57+
58+
return {
59+
"engineType": engine_type,
60+
"engineClass": engine_class,
61+
"config": engine_config,
62+
}
63+
64+
65+
def _extract_materialization_jobs(
66+
grpc_handler, project: str, allow_cache: bool = True
67+
) -> List[Dict[str, Any]]:
68+
"""Extract materialization job info from feature view metadata."""
69+
req = RegistryServer_pb2.ListFeatureViewsRequest(
70+
project=project,
71+
allow_cache=allow_cache,
72+
)
73+
response = grpc_call(grpc_handler.ListFeatureViews, req)
74+
feature_views = response.get("featureViews", [])
75+
76+
jobs: List[Dict[str, Any]] = []
77+
for fv in feature_views:
78+
spec = fv.get("spec", fv)
79+
meta = fv.get("meta", {})
80+
fv_name = spec.get("name", "unknown")
81+
intervals = meta.get("materializationIntervals", [])
82+
83+
for idx, interval in enumerate(intervals):
84+
start_time = interval.get("startTime", interval.get("start_time"))
85+
end_time = interval.get("endTime", interval.get("end_time"))
86+
87+
jobs.append(
88+
{
89+
"id": f"{fv_name}-{idx}",
90+
"featureView": fv_name,
91+
"status": "SUCCEEDED",
92+
"startTime": start_time,
93+
"endTime": end_time,
94+
}
95+
)
96+
97+
jobs.sort(key=lambda j: j.get("startTime", ""), reverse=True)
98+
return jobs
99+
100+
101+
def get_compute_engine_router(grpc_handler, store=None) -> APIRouter:
102+
router = APIRouter()
103+
104+
def _fv_to_info(fv: dict, fv_type: str) -> Dict[str, Any]:
105+
spec = fv.get("spec", fv)
106+
meta = fv.get("meta", {})
107+
intervals = meta.get("materializationIntervals", [])
108+
batch_engine = spec.get("batchEngine")
109+
110+
last_materialized = None
111+
if intervals:
112+
last_interval = intervals[-1]
113+
last_materialized = last_interval.get(
114+
"endTime", last_interval.get("end_time")
115+
)
116+
117+
return {
118+
"name": spec.get("name"),
119+
"type": fv_type,
120+
"online": spec.get("online", True),
121+
"lastMaterialized": last_materialized,
122+
"hasOverride": batch_engine is not None,
123+
"overrides": batch_engine,
124+
"materializationIntervals": intervals,
125+
}
126+
127+
@router.get("/compute_engines")
128+
def get_compute_engine(
129+
project: str = Query(...),
130+
allow_cache: bool = Query(default=True),
131+
):
132+
engine_info = (
133+
_extract_engine_info(store)
134+
if store
135+
else {
136+
"engineType": "local",
137+
"engineClass": "LocalComputeEngine",
138+
"config": {"type": "local"},
139+
}
140+
)
141+
142+
fv_infos: List[Dict[str, Any]] = []
143+
144+
batch_resp = grpc_call(
145+
grpc_handler.ListFeatureViews,
146+
RegistryServer_pb2.ListFeatureViewsRequest(
147+
project=project, allow_cache=allow_cache
148+
),
149+
)
150+
for fv in batch_resp.get("featureViews", []):
151+
fv_infos.append(_fv_to_info(fv, "Batch"))
152+
153+
stream_resp = grpc_call(
154+
grpc_handler.ListStreamFeatureViews,
155+
RegistryServer_pb2.ListStreamFeatureViewsRequest(
156+
project=project, allow_cache=allow_cache
157+
),
158+
)
159+
for sfv in stream_resp.get("streamFeatureViews", []):
160+
fv_infos.append(_fv_to_info(sfv, "Stream"))
161+
162+
engine_info["featureViewCount"] = len(fv_infos)
163+
engine_info["project"] = project
164+
165+
return {
166+
"engine": engine_info,
167+
"featureViews": fv_infos,
168+
}
169+
170+
@router.get("/compute_engines/all")
171+
def list_all_compute_engines(
172+
allow_cache: bool = Query(default=True),
173+
page: int = Query(1, ge=1),
174+
limit: int = Query(50, ge=1, le=100),
175+
):
176+
req = RegistryServer_pb2.ListProjectsRequest(allow_cache=allow_cache)
177+
projects_resp = grpc_call(grpc_handler.ListProjects, req)
178+
projects = projects_resp.get("projects", [])
179+
180+
engines = []
181+
for proj in projects:
182+
proj_name = proj.get("spec", {}).get("name") or proj.get("name", "")
183+
if not proj_name:
184+
continue
185+
186+
engine_info = (
187+
_extract_engine_info(store)
188+
if store
189+
else {
190+
"engineType": "local",
191+
"engineClass": "LocalComputeEngine",
192+
"config": {"type": "local"},
193+
"mode": "standalone",
194+
"configSource": "feature_store.yaml",
195+
}
196+
)
197+
198+
fv_req = RegistryServer_pb2.ListFeatureViewsRequest(
199+
project=proj_name,
200+
allow_cache=allow_cache,
201+
)
202+
fv_resp = grpc_call(grpc_handler.ListFeatureViews, fv_req)
203+
fv_count = len(fv_resp.get("featureViews", []))
204+
205+
engine_info["project"] = proj_name
206+
engine_info["featureViewCount"] = fv_count
207+
engines.append(engine_info)
208+
209+
start = (page - 1) * limit
210+
end = start + limit
211+
paginated = engines[start:end]
212+
213+
return {
214+
"engines": paginated,
215+
"pagination": {
216+
"page": page,
217+
"limit": limit,
218+
"total": len(engines),
219+
},
220+
}
221+
222+
@router.get("/materialization_jobs")
223+
def list_materialization_jobs(
224+
project: str = Query(...),
225+
allow_cache: bool = Query(default=True),
226+
status: Optional[str] = Query(None, description="Filter by status"),
227+
feature_view: Optional[str] = Query(
228+
None, alias="feature_view", description="Filter by feature view name"
229+
),
230+
pagination_params: dict = Depends(get_pagination_params),
231+
sorting_params: dict = Depends(get_sorting_params),
232+
):
233+
jobs = _extract_materialization_jobs(grpc_handler, project, allow_cache)
234+
235+
if status:
236+
jobs = [j for j in jobs if j.get("status") == status.upper()]
237+
if feature_view:
238+
jobs = [j for j in jobs if j.get("featureView") == feature_view]
239+
240+
page = pagination_params.get("page", 1)
241+
limit = pagination_params.get("limit", 50)
242+
start = (page - 1) * limit
243+
end = start + limit
244+
total = len(jobs)
245+
paginated = jobs[start:end]
246+
247+
return {
248+
"jobs": paginated,
249+
"pagination": {
250+
"page": page,
251+
"limit": limit,
252+
"total": total,
253+
},
254+
}
255+
256+
return router

‎sdk/python/tests/unit/api/test_api_rest_registry_server.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -62,4 +62,4 @@ def test_routes_registered_in_app():
6262
server = MagicMock()
6363
register_all_routes(app, grpc_handler, server)
6464

65-
assert app.include_router.call_count == 13
65+
assert app.include_router.call_count == 14

‎ui/src/FeastUISansProviders.tsx‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@ import LineageIndex from "./pages/lineage/Index";
2929
import NoProjectGuard from "./components/NoProjectGuard";
3030
import MonitoringIndex from "./pages/monitoring/Index";
3131
import FeatureMetricsDetail from "./pages/monitoring/FeatureMetricsDetail";
32+
import ComputeEngineIndex from "./pages/compute-engines/Index";
3233

3334
import TabsRegistryContext, {
3435
FeastTabsRegistryInterface,
@@ -224,6 +225,10 @@ const FeastUISansProvidersInner = ({
224225
path="monitoring/feature/:featureViewName/:featureName"
225226
element={<FeatureMetricsDetail />}
226227
/>
228+
<Route
229+
path="compute-engine/*"
230+
element={<ComputeEngineIndex />}
231+
/>
227232
</Route>
228233
</Route>
229234
<Route path="*" element={<NoMatch />} />

‎ui/src/graphics/ComputeEngineIcon.tsx‎

Lines changed: 41 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,41 @@
1+
import React from "react";
2+
3+
const ComputeEngineIcon = (props: React.SVGProps<SVGSVGElement>) => {
4+
return (
5+
<svg {...props} viewBox="0 0 32 32" fill="none">
6+
<path
7+
d="M4 8C4 6.89543 4.89543 6 6 6H26C27.1046 6 28 6.89543 28 8V12C28 13.1046 27.1046 14 26 14H6C4.89543 14 4 13.1046 4 12V8Z"
8+
fill="#0569EA"
9+
/>
10+
<circle cx="8" cy="10" r="1.5" fill="white" />
11+
<circle cx="12" cy="10" r="1.5" fill="white" />
12+
<rect
13+
x="16"
14+
y="9"
15+
width="8"
16+
height="2"
17+
rx="1"
18+
fill="white"
19+
opacity="0.6"
20+
/>
21+
<path
22+
d="M4 20C4 18.8954 4.89543 18 6 18H26C27.1046 18 28 18.8954 28 20V24C28 25.1046 27.1046 26 26 26H6C4.89543 26 4 25.1046 4 24V20Z"
23+
fill="#0569EA"
24+
opacity="0.6"
25+
/>
26+
<circle cx="8" cy="22" r="1.5" fill="white" />
27+
<circle cx="12" cy="22" r="1.5" fill="white" />
28+
<rect
29+
x="16"
30+
y="21"
31+
width="8"
32+
height="2"
33+
rx="1"
34+
fill="white"
35+
opacity="0.6"
36+
/>
37+
</svg>
38+
);
39+
};
40+
41+
export { ComputeEngineIcon };

‎ui/src/graphics/JobsIcon.tsx‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,18 @@
1+
import React from "react";
2+
3+
const JobsIcon = (props: React.SVGProps<SVGSVGElement>) => {
4+
return (
5+
<svg {...props} viewBox="0 0 32 32" fill="none">
6+
<path
7+
d="M16 4C9.37258 4 4 9.37258 4 16C4 22.6274 9.37258 28 16 28C22.6274 28 28 22.6274 28 16C28 9.37258 22.6274 4 16 4ZM16 6.5C21.2467 6.5 25.5 10.7533 25.5 16C25.5 21.2467 21.2467 25.5 16 25.5C10.7533 25.5 6.5 21.2467 6.5 16C6.5 10.7533 10.7533 6.5 16 6.5Z"
8+
fill="#0569EA"
9+
/>
10+
<path
11+
d="M15 10V16.414L19.293 20.707L20.707 19.293L17 15.586V10H15Z"
12+
fill="#0569EA"
13+
/>
14+
</svg>
15+
);
16+
};
17+
18+
export { JobsIcon };

‎ui/src/pages/Sidebar.tsx‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@ import { FeatureIcon } from "../graphics/FeatureIcon";
2424
import { HomeIcon } from "../graphics/HomeIcon";
2525
import { PermissionsIcon } from "../graphics/PermissionsIcon";
2626
import { LabelViewIcon } from "../graphics/LabelViewIcon";
27+
import { ComputeEngineIcon } from "../graphics/ComputeEngineIcon";
2728
import type { genericFVType } from "../parsers/mergedFVTypes";
2829

2930
const SideNav = () => {
@@ -195,6 +196,15 @@ const SideNav = () => {
195196
),
196197
isSelected: monitoringSelected,
197198
},
199+
{
200+
name: "Compute & Jobs",
201+
id: htmlIdGenerator("computeEngine")(),
202+
icon: <EuiIcon type={ComputeEngineIcon} />,
203+
renderItem: (props: any) => (
204+
<Link {...props} to={`${baseUrl}/compute-engine`} />
205+
),
206+
isSelected: useMatchSubpath(`${baseUrl}/compute-engine`),
207+
},
198208
],
199209
},
200210
];

0 commit comments

Comments
 (0)