From 3a054ac13dd5424cc52a4a9e9e8b3f6b5dea269f Mon Sep 17 00:00:00 2001 From: mo Date: Sun, 28 Jun 2026 13:23:27 +0000 Subject: [PATCH] Elasticsearch: full indexer, rich search UI, Kibana dashboards + shortcut MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Backend: Trino-federated indexer pushes all sources (postgres/mysql/mongo/cassandra/iceberg/neo4j + catalog) into atc-* indices; adds /indices, /mapping, /query, /aggs, /reindex(+status), /kibana/setup(+links) - Kibana: auto-provision data views + ATC Data Overview dashboard (docs by source, top tables) - UI: rebuilt Search tab — KPIs, source/index charts, full-text + filtered search, facets, paginated doc viewer, re-index + Kibana buttons - SideNav: Kibana Dashboards shortcut --- api/elasticsearch_api.py | 578 ++++++++++++++++++++-- ui/src/components/features/SearchView.tsx | 498 ++++++++++++++++--- ui/src/components/layout/SideNav.tsx | 22 +- 3 files changed, 976 insertions(+), 122 deletions(-) diff --git a/api/elasticsearch_api.py b/api/elasticsearch_api.py index 9ac08e5..515b67d 100644 --- a/api/elasticsearch_api.py +++ b/api/elasticsearch_api.py @@ -1,12 +1,24 @@ -"""Elasticsearch + Kibana health API for Command Center.""" +"""Elasticsearch + Kibana API for the Command Center. + +Provides: + - cluster / kibana health + - rich index listing, field mappings, full-text + filtered search, aggregations + - a Trino-federated indexer that pushes every source's data into ES (atc-* indices) + - Kibana provisioning (data views + an overview dashboard) +""" from __future__ import annotations +import json import os +import re +import threading +import time +from datetime import datetime, timezone from typing import Any import httpx -from fastapi import APIRouter, Query +from fastapi import APIRouter, Body, Query from fastapi.responses import JSONResponse ELASTICSEARCH_URL = os.getenv("ELASTICSEARCH_URL", "https://10.0.21.46:9200").rstrip("/") @@ -14,21 +26,48 @@ KIBANA_URL = os.getenv("KIBANA_URL", "http://10.0.21.46:5601").rstrip("/") ELASTIC_USER = os.getenv("ELASTIC_USER", "elastic") ELASTIC_PASSWORD = os.getenv("ELASTIC_PASSWORD", "") +INDEX_PREFIX = "atc-" +# Trino catalogs to index (skip system + streaming kafka). +INDEX_CATALOGS = ("postgres_sales", "mysql_hr", "mongodb_supplychain", "cassandra_telemetry", "iceberg") +SKIP_SCHEMAS = {"information_schema", "sys", "performance_schema", "config", "local", "admin"} +ROWS_PER_TABLE = int(os.getenv("ES_ROWS_PER_TABLE", "500")) + router = APIRouter(prefix="/api/search", tags=["search"]) +def _auth(): + return (ELASTIC_USER, ELASTIC_PASSWORD) if ELASTIC_PASSWORD else None + + +def _es_client(timeout: float = 30.0) -> httpx.Client: + return httpx.Client(timeout=timeout, verify=False, auth=_auth()) + + +def _sanitize_index(*parts: str) -> str: + raw = "-".join(str(p) for p in parts if p) + raw = raw.lower() + raw = re.sub(r"[^a-z0-9]+", "-", raw).strip("-") + raw = re.sub(r"-+", "-", raw) + return (INDEX_PREFIX + raw)[:240] + + +# ────────────────────────────────────────────────────────────────────────────── +# Health +# ────────────────────────────────────────────────────────────────────────────── async def _probe_es() -> dict[str, Any]: - auth = (ELASTIC_USER, ELASTIC_PASSWORD) if ELASTIC_PASSWORD else None try: async with httpx.AsyncClient(timeout=6.0, verify=False) as client: - r = await client.get(f"{ELASTICSEARCH_URL}/", auth=auth) + r = await client.get(f"{ELASTICSEARCH_URL}/", auth=_auth()) if r.status_code >= 400: return {"ok": False, "status_code": r.status_code, "error": r.text[:200]} info = r.json() - health_r = await client.get(f"{ELASTICSEARCH_URL}/_cluster/health", auth=auth) + health_r = await client.get(f"{ELASTICSEARCH_URL}/_cluster/health", auth=_auth()) health = health_r.json() if health_r.status_code < 400 else {} - stats_r = await client.get(f"{ELASTICSEARCH_URL}/_cat/indices?format=json&bytes=b", auth=auth) + stats_r = await client.get(f"{ELASTICSEARCH_URL}/_cat/indices?format=json&bytes=b", auth=_auth()) indices = stats_r.json() if stats_r.status_code < 400 else [] + user_idx = [i for i in indices if isinstance(i, dict) and not str(i.get("index", "")).startswith(".")] + total_docs = sum(int(i.get("docs.count") or 0) for i in user_idx) + atc_docs = sum(int(i.get("docs.count") or 0) for i in user_idx if str(i.get("index", "")).startswith(INDEX_PREFIX)) return { "ok": True, "url": ELASTICSEARCH_URL, @@ -36,7 +75,9 @@ async def _probe_es() -> dict[str, Any]: "version": info.get("version", {}).get("number"), "health": health.get("status", "unknown"), "nodes": health.get("number_of_nodes"), - "indices_count": len(indices) if isinstance(indices, list) else 0, + "indices_count": len(user_idx), + "total_docs": total_docs, + "atc_docs": atc_docs, "indices": [ { "name": i.get("index"), @@ -44,10 +85,10 @@ async def _probe_es() -> dict[str, Any]: "size": i.get("store.size"), "health": i.get("health"), } - for i in (indices[:50] if isinstance(indices, list) else []) + for i in user_idx[:80] ], } - except Exception as exc: + except Exception as exc: # noqa: BLE001 return {"ok": False, "url": ELASTICSEARCH_URL, "error": str(exc)} @@ -56,27 +97,24 @@ async def _probe_kibana() -> dict[str, Any]: async with httpx.AsyncClient(timeout=6.0, verify=False) as client: r = await client.get(f"{KIBANA_URL}/api/status") if r.status_code >= 400: - return {"ok": False, "status_code": r.status_code} + return {"ok": False, "url": KIBANA_URL, "ui_url": KIBANA_URL, "status_code": r.status_code} data = r.json() overall = data.get("status", {}).get("overall", {}) return { - "ok": overall.get("level") in ("available", "green", "yellow"), + "ok": overall.get("level", "available") in ("available", "green", "yellow"), "url": KIBANA_URL, "ui_url": KIBANA_URL, - "level": overall.get("level", "unknown"), + "level": overall.get("level", "available"), + "version": data.get("version", {}).get("number"), } - except Exception as exc: - return {"ok": False, "url": KIBANA_URL, "error": str(exc)} + except Exception as exc: # noqa: BLE001 + return {"ok": False, "url": KIBANA_URL, "ui_url": KIBANA_URL, "error": str(exc)} @router.get("/health") async def search_health(): es, kb = await _probe_es(), await _probe_kibana() - return { - "ok": es.get("ok") or kb.get("ok"), - "elasticsearch": es, - "kibana": kb, - } + return {"ok": es.get("ok") or kb.get("ok"), "elasticsearch": es, "kibana": kb} @router.get("/elasticsearch") @@ -92,29 +130,491 @@ async def get_kibana(): return await _probe_kibana() -@router.get("/elasticsearch/query") -async def es_search(q: str = Query(..., min_length=1), size: int = Query(10, le=50)): - if not ELASTIC_PASSWORD: - return JSONResponse({"ok": False, "error": "ELASTIC_PASSWORD not configured"}, status_code=503) +# ────────────────────────────────────────────────────────────────────────────── +# Indices + mappings +# ────────────────────────────────────────────────────────────────────────────── +@router.get("/indices") +async def list_indices(include_system: bool = False): + try: + async with httpx.AsyncClient(timeout=10.0, verify=False) as client: + r = await client.get(f"{ELASTICSEARCH_URL}/_cat/indices?format=json&bytes=b&s=index", auth=_auth()) + if r.status_code >= 400: + return JSONResponse({"ok": False, "error": r.text[:300]}, status_code=502) + rows = r.json() + except Exception as exc: # noqa: BLE001 + return JSONResponse({"ok": False, "error": str(exc)}, status_code=502) + + out = [] + for i in rows: + name = str(i.get("index", "")) + if not include_system and name.startswith("."): + continue + out.append({ + "name": name, + "docs": int(i.get("docs.count") or 0), + "size_bytes": int(i.get("store.size") or 0), + "health": i.get("health"), + "status": i.get("status"), + "atc": name.startswith(INDEX_PREFIX), + }) + out.sort(key=lambda x: (not x["atc"], -x["docs"])) + return {"ok": True, "count": len(out), "indices": out} + + +def _flatten_props(props: dict, prefix: str = "") -> list[dict]: + fields = [] + for name, spec in (props or {}).items(): + full = f"{prefix}{name}" + ftype = spec.get("type") + if ftype: + aggregatable = ftype in ("keyword", "long", "integer", "double", "float", "date", "boolean", "ip") + fields.append({"name": full, "type": ftype, "aggregatable": aggregatable}) + sub = spec.get("fields") or {} + for sub_name, sub_spec in sub.items(): + st = sub_spec.get("type") + fields.append({"name": f"{full}.{sub_name}", "type": st, "aggregatable": st in ("keyword", "ip")}) + if spec.get("properties"): + fields.extend(_flatten_props(spec["properties"], prefix=f"{full}.")) + return fields + + +@router.get("/mapping") +async def get_mapping(index: str = Query(...)): + try: + async with httpx.AsyncClient(timeout=10.0, verify=False) as client: + r = await client.get(f"{ELASTICSEARCH_URL}/{index}/_mapping", auth=_auth()) + if r.status_code >= 400: + return JSONResponse({"ok": False, "error": r.text[:300]}, status_code=502) + data = r.json() + except Exception as exc: # noqa: BLE001 + return JSONResponse({"ok": False, "error": str(exc)}, status_code=502) + + fields: dict[str, dict] = {} + for _idx, body in data.items(): + props = body.get("mappings", {}).get("properties", {}) + for f in _flatten_props(props): + fields[f["name"]] = f + field_list = sorted(fields.values(), key=lambda x: x["name"]) + return {"ok": True, "index": index, "fields": field_list} + + +# ────────────────────────────────────────────────────────────────────────────── +# Search + aggregations +# ────────────────────────────────────────────────────────────────────────────── +@router.post("/query") +async def search_query(body: dict = Body(default={})): + q = (body.get("q") or "").strip() + indices = body.get("indices") or [] + filters = body.get("filters") or [] + from_ = int(body.get("from") or 0) + size = min(int(body.get("size") or 20), 100) + sort = body.get("sort") + + target = ",".join(indices) if indices else f"{INDEX_PREFIX}*" + + must: list[dict] = [] + if q: + must.append({"query_string": {"query": q, "lenient": True, "default_operator": "AND"}}) + for f in filters: + field, value = f.get("field"), f.get("value") + if field and value is not None: + must.append({"match_phrase": {field: value}}) + es_body: dict[str, Any] = { + "query": {"bool": {"must": must or [{"match_all": {}}]}}, + "from": from_, + "size": size, + "track_total_hits": True, + } + if sort and sort.get("field"): + es_body["sort"] = [{sort["field"]: {"order": sort.get("order", "desc")}}] + + try: + async with httpx.AsyncClient(timeout=20.0, verify=False) as client: + r = await client.post(f"{ELASTICSEARCH_URL}/{target}/_search", auth=_auth(), json=es_body) + if r.status_code >= 400: + return JSONResponse({"ok": False, "error": r.text[:400]}, status_code=r.status_code) + data = r.json() + except Exception as exc: # noqa: BLE001 + return JSONResponse({"ok": False, "error": str(exc)}, status_code=502) + + hits = data.get("hits", {}) + return { + "ok": True, + "took": data.get("took"), + "total": hits.get("total", {}).get("value", 0), + "hits": [ + {"index": h.get("_index"), "id": h.get("_id"), "score": h.get("_score"), "source": h.get("_source")} + for h in hits.get("hits", []) + ], + } + + +@router.get("/aggs") +async def aggregations(index: str = Query(...), field: str = Query(...), size: int = Query(15, le=50), q: str = Query("")): + query = {"query_string": {"query": q, "lenient": True}} if q.strip() else {"match_all": {}} + es_body = {"size": 0, "query": query, "aggs": {"facet": {"terms": {"field": field, "size": size}}}} try: async with httpx.AsyncClient(timeout=15.0, verify=False) as client: - r = await client.post( - f"{ELASTICSEARCH_URL}/_search", - auth=(ELASTIC_USER, ELASTIC_PASSWORD), - json={"query": {"query_string": {"query": q}}, "size": size}, - ) + r = await client.post(f"{ELASTICSEARCH_URL}/{index}/_search", auth=_auth(), json=es_body) if r.status_code >= 400: return JSONResponse({"ok": False, "error": r.text[:300]}, status_code=r.status_code) data = r.json() - hits = [ - { - "index": h.get("_index"), - "id": h.get("_id"), - "score": h.get("_score"), - "source": h.get("_source"), - } - for h in data.get("hits", {}).get("hits", []) - ] - return {"ok": True, "total": data.get("hits", {}).get("total", {}), "hits": hits} - except Exception as exc: + except Exception as exc: # noqa: BLE001 return JSONResponse({"ok": False, "error": str(exc)}, status_code=502) + buckets = data.get("aggregations", {}).get("facet", {}).get("buckets", []) + return {"ok": True, "field": field, "buckets": [{"key": b.get("key"), "count": b.get("doc_count")} for b in buckets]} + + +# back-compat simple query +@router.get("/elasticsearch/query") +async def es_search(q: str = Query(..., min_length=1), size: int = Query(10, le=50)): + return await search_query({"q": q, "size": size}) + + +# ────────────────────────────────────────────────────────────────────────────── +# Indexer (Trino-federated → ES atc-* indices) +# ────────────────────────────────────────────────────────────────────────────── +_reindex_state: dict[str, Any] = { + "running": False, + "started_at": None, + "finished_at": None, + "current": None, + "indices": {}, + "total_docs": 0, + "errors": [], + "log": [], +} +_reindex_lock = threading.Lock() + + +def _bulk(client: httpx.Client, index: str, docs: list[tuple[str, dict]]) -> int: + if not docs: + return 0 + lines = [] + for _id, doc in docs: + lines.append(json.dumps({"index": {"_index": index, "_id": _id}})) + lines.append(json.dumps(doc, default=str)) + body = "\n".join(lines) + "\n" + r = client.post( + f"{ELASTICSEARCH_URL}/_bulk", + content=body.encode("utf-8"), + headers={"Content-Type": "application/x-ndjson"}, + ) + if r.status_code >= 400: + raise RuntimeError(f"bulk failed {r.status_code}: {r.text[:200]}") + return len(docs) + + +def _meta(catalog: str, schema: str, table: str, source: str) -> dict: + return { + "catalog": catalog, + "schema": schema, + "table": table, + "fqn": f"{catalog}.{schema}.{table}" if schema else f"{catalog}.{table}", + "source": source, + "indexed_at": datetime.now(timezone.utc).isoformat(), + } + + +def _index_trino_table(sqlmod, client: httpx.Client, catalog: str, schema: str, table: str) -> int: + res = sqlmod._run_trino(f'SELECT * FROM "{catalog}"."{schema}"."{table}" LIMIT {ROWS_PER_TABLE}', ROWS_PER_TABLE) + if not res.get("ok"): + raise RuntimeError(res.get("error", "query failed")[:200]) + cols = res.get("columns", []) + rows = res.get("rows", []) + index = _sanitize_index(catalog, schema, table) + docs = [] + for n, row in enumerate(rows): + doc = {cols[i]: row[i] for i in range(min(len(cols), len(row)))} + doc["meta"] = _meta(catalog, schema, table, catalog) + docs.append((f"{catalog}.{schema}.{table}:{n}", doc)) + written = 0 + for i in range(0, len(docs), 500): + written += _bulk(client, index, docs[i:i + 500]) + return written + + +def _index_neo4j(sqlmod, client: httpx.Client) -> int: + written = 0 + try: + driver = sqlmod._neo4j_driver() + except Exception: # noqa: BLE001 + return 0 + try: + with driver.session() as session: + labels = [r["label"] for r in session.run("CALL db.labels() YIELD label RETURN label")] + for label in labels: + index = _sanitize_index("neo4j", "graph", label) + recs = session.run(f"MATCH (n:`{label}`) RETURN properties(n) AS props LIMIT {ROWS_PER_TABLE}") + docs = [] + for n, rec in enumerate(recs): + props = dict(rec["props"] or {}) + props["meta"] = _meta("neo4j", "graph", label, "neo4j") + docs.append((f"neo4j.{label}:{n}", props)) + for i in range(0, len(docs), 500): + written += _bulk(client, index, docs[i:i + 500]) + _reindex_state["indices"][index] = len(docs) + _reindex_state["log"].append(f"neo4j:{label} → {len(docs)} docs") + except Exception as exc: # noqa: BLE001 + _reindex_state["errors"].append(f"neo4j: {exc}") + finally: + driver.close() + return written + + +def _index_catalog(client: httpx.Client) -> int: + try: + from database_inventory import collect_database_inventory_sync + inv = collect_database_inventory_sync() + except Exception as exc: # noqa: BLE001 + _reindex_state["errors"].append(f"catalog: {exc}") + return 0 + docs = [] + for engine, info in (inv.get("engines") or {}).items(): + for obj in (info.get("objects") or info.get("tables") or []): + if not isinstance(obj, dict): + continue + doc = dict(obj) + doc["engine"] = engine + doc["meta"] = _meta("catalog", engine, str(obj.get("name") or obj.get("fqn") or "obj"), "inventory") + docs.append((f"catalog.{engine}.{obj.get('fqn') or obj.get('name')}", doc)) + written = 0 + for i in range(0, len(docs), 500): + written += _bulk(client, "atc-catalog", docs[i:i + 500]) + if docs: + _reindex_state["indices"]["atc-catalog"] = len(docs) + _reindex_state["log"].append(f"catalog → {len(docs)} docs") + return written + + +def _reindex_worker(): + import sql_console as sqlmod + try: + client = _es_client(timeout=90.0) + with client: + # catalog/metadata + _reindex_state["current"] = "catalog inventory" + _reindex_state["total_docs"] += _index_catalog(client) + + # Trino-federated source data + cat_res = sqlmod._run_trino("SHOW CATALOGS", 100) + catalogs = [r[0] for r in cat_res.get("rows", [])] if cat_res.get("ok") else list(INDEX_CATALOGS) + for catalog in catalogs: + if catalog not in INDEX_CATALOGS: + continue + try: + sch_res = sqlmod._run_trino(f'SHOW SCHEMAS FROM "{catalog}"', 200) + schemas = [r[0] for r in sch_res.get("rows", [])] if sch_res.get("ok") else [] + except Exception as exc: # noqa: BLE001 + _reindex_state["errors"].append(f"{catalog}: {exc}") + continue + for schema in schemas: + if schema in SKIP_SCHEMAS or schema.startswith("system") or schema == "information_schema": + continue + try: + tbl_res = sqlmod._run_trino(f'SHOW TABLES FROM "{catalog}"."{schema}"', 500) + tables = [r[0] for r in tbl_res.get("rows", [])] if tbl_res.get("ok") else [] + except Exception as exc: # noqa: BLE001 + _reindex_state["errors"].append(f"{catalog}.{schema}: {exc}") + continue + for table in tables: + _reindex_state["current"] = f"{catalog}.{schema}.{table}" + try: + n = _index_trino_table(sqlmod, client, catalog, schema, table) + idx = _sanitize_index(catalog, schema, table) + _reindex_state["indices"][idx] = n + _reindex_state["total_docs"] += n + _reindex_state["log"].append(f"{catalog}.{schema}.{table} → {n} docs") + except Exception as exc: # noqa: BLE001 + _reindex_state["errors"].append(f"{catalog}.{schema}.{table}: {str(exc)[:150]}") + + # Neo4j (not federated by Trino) + _reindex_state["current"] = "neo4j graph" + _reindex_state["total_docs"] += _index_neo4j(sqlmod, client) + + # make freshly created indices searchable immediately + try: + client.post(f"{ELASTICSEARCH_URL}/{INDEX_PREFIX}*/_refresh") + except Exception: # noqa: BLE001 + pass + except Exception as exc: # noqa: BLE001 + _reindex_state["errors"].append(f"fatal: {exc}") + finally: + _reindex_state["current"] = None + _reindex_state["running"] = False + _reindex_state["finished_at"] = datetime.now(timezone.utc).isoformat() + # keep log bounded + _reindex_state["log"] = _reindex_state["log"][-200:] + + +@router.post("/reindex") +async def start_reindex(): + with _reindex_lock: + if _reindex_state["running"]: + return {"ok": True, "already_running": True, "state": _reindex_summary()} + _reindex_state.update({ + "running": True, + "started_at": datetime.now(timezone.utc).isoformat(), + "finished_at": None, + "current": "starting…", + "indices": {}, + "total_docs": 0, + "errors": [], + "log": [], + }) + threading.Thread(target=_reindex_worker, daemon=True).start() + return {"ok": True, "started": True} + + +def _reindex_summary() -> dict[str, Any]: + s = _reindex_state + return { + "running": s["running"], + "started_at": s["started_at"], + "finished_at": s["finished_at"], + "current": s["current"], + "total_docs": s["total_docs"], + "index_count": len(s["indices"]), + "indices": s["indices"], + "errors": s["errors"][-20:], + "log": s["log"][-40:], + } + + +@router.get("/reindex/status") +async def reindex_status(): + return {"ok": True, **_reindex_summary()} + + +# ────────────────────────────────────────────────────────────────────────────── +# Kibana provisioning +# ────────────────────────────────────────────────────────────────────────────── +def _kbn_headers(): + return {"kbn-xsrf": "true", "Content-Type": "application/json"} + + +@router.post("/kibana/setup") +async def kibana_setup(): + """Create data views + an overview dashboard in Kibana (idempotent).""" + created: list[str] = [] + errors: list[str] = [] + dataview_id = "atc-all-data" + objects = [ + { + "type": "index-pattern", + "id": dataview_id, + "attributes": {"title": f"{INDEX_PREFIX}*", "name": "ATC — all data", "timeFieldName": "meta.indexed_at"}, + }, + { + "type": "index-pattern", + "id": "atc-catalog-dv", + "attributes": {"title": "atc-catalog", "name": "ATC — data catalog"}, + }, + { + "type": "visualization", + "id": "atc-docs-by-source", + "attributes": { + "title": "ATC — Documents by source", + "visState": json.dumps({ + "title": "ATC — Documents by source", + "type": "histogram", + "aggs": [ + {"id": "1", "enabled": True, "type": "count", "schema": "metric", "params": {}}, + {"id": "2", "enabled": True, "type": "terms", "schema": "segment", + "params": {"field": "meta.catalog.keyword", "size": 25, "order": "desc", "orderBy": "1"}}, + ], + "params": {"addLegend": True, "addTooltip": True, "type": "histogram"}, + }), + "uiStateJSON": "{}", + "description": "", + "kibanaSavedObjectMeta": {"searchSourceJSON": json.dumps({ + "query": {"query": "", "language": "kuery"}, "filter": [], + "indexRefName": "kibanaSavedObjectMeta.searchSourceJSON.index", + })}, + }, + "references": [{"name": "kibanaSavedObjectMeta.searchSourceJSON.index", "type": "index-pattern", "id": dataview_id}], + }, + { + "type": "visualization", + "id": "atc-docs-by-table", + "attributes": { + "title": "ATC — Top tables", + "visState": json.dumps({ + "title": "ATC — Top tables", + "type": "table", + "aggs": [ + {"id": "1", "enabled": True, "type": "count", "schema": "metric", "params": {}}, + {"id": "2", "enabled": True, "type": "terms", "schema": "bucket", + "params": {"field": "meta.table.keyword", "size": 30, "order": "desc", "orderBy": "1"}}, + ], + "params": {"perPage": 15, "showPartialRows": False, "showTotal": True}, + }), + "uiStateJSON": "{}", + "description": "", + "kibanaSavedObjectMeta": {"searchSourceJSON": json.dumps({ + "query": {"query": "", "language": "kuery"}, "filter": [], + "indexRefName": "kibanaSavedObjectMeta.searchSourceJSON.index", + })}, + }, + "references": [{"name": "kibanaSavedObjectMeta.searchSourceJSON.index", "type": "index-pattern", "id": dataview_id}], + }, + { + "type": "dashboard", + "id": "atc-data-overview", + "attributes": { + "title": "ATC Data Overview", + "description": "Auto-provisioned overview of all data indexed from the ATC platform.", + "panelsJSON": json.dumps([ + {"version": "8.19.0", "type": "visualization", "panelIndex": "1", "panelRefName": "panel_1", + "gridData": {"x": 0, "y": 0, "w": 24, "h": 15, "i": "1"}, "embeddableConfig": {}}, + {"version": "8.19.0", "type": "visualization", "panelIndex": "2", "panelRefName": "panel_2", + "gridData": {"x": 24, "y": 0, "w": 24, "h": 15, "i": "2"}, "embeddableConfig": {}}, + ]), + "optionsJSON": json.dumps({"useMargins": True, "hidePanelTitles": False}), + "timeRestore": False, + "kibanaSavedObjectMeta": {"searchSourceJSON": json.dumps({"query": {"query": "", "language": "kuery"}, "filter": []})}, + }, + "references": [ + {"name": "panel_1", "type": "visualization", "id": "atc-docs-by-source"}, + {"name": "panel_2", "type": "visualization", "id": "atc-docs-by-table"}, + ], + }, + ] + + try: + async with httpx.AsyncClient(timeout=20.0, verify=False, auth=_auth()) as client: + r = await client.post( + f"{KIBANA_URL}/api/saved_objects/_bulk_create?overwrite=true", + headers=_kbn_headers(), + json=objects, + ) + if r.status_code >= 400: + return JSONResponse({"ok": False, "error": r.text[:400]}, status_code=r.status_code) + data = r.json() + for obj in data.get("saved_objects", []): + if obj.get("error"): + errors.append(f"{obj.get('type')}/{obj.get('id')}: {obj['error'].get('message')}") + else: + created.append(f"{obj.get('type')}/{obj.get('id')}") + except Exception as exc: # noqa: BLE001 + return JSONResponse({"ok": False, "error": str(exc)}, status_code=502) + + return { + "ok": True, + "created": created, + "errors": errors, + "dashboard_url": f"{KIBANA_URL}/app/dashboards#/view/atc-data-overview", + "discover_url": f"{KIBANA_URL}/app/discover", + } + + +@router.get("/kibana/links") +async def kibana_links(index: str = Query("")): + base = KIBANA_URL + return { + "ok": True, + "kibana": base, + "discover": f"{base}/app/discover", + "dashboard": f"{base}/app/dashboards#/view/atc-data-overview", + "index_management": f"{base}/app/management/data/index_management/indices", + } diff --git a/ui/src/components/features/SearchView.tsx b/ui/src/components/features/SearchView.tsx index c2ff137..1f70807 100644 --- a/ui/src/components/features/SearchView.tsx +++ b/ui/src/components/features/SearchView.tsx @@ -1,5 +1,8 @@ -import { useCallback, useEffect, useState } from 'react' -import { ExternalLink, Loader2, RefreshCw, Search } from 'lucide-react' +import { useCallback, useEffect, useMemo, useRef, useState } from 'react' +import { + BarChart3, ChevronDown, ChevronRight, Database, ExternalLink, Filter, LayoutDashboard, + Layers, Loader2, Play, RefreshCw, Search, Sparkles, X, +} from 'lucide-react' import { cn } from '../../lib/utils' import { subTabActive, subTabIdle } from '../../lib/tabActive' @@ -10,20 +13,87 @@ type EsHealth = { health?: string nodes?: number indices_count?: number - indices?: { name?: string; docs?: string; size?: string; health?: string }[] + total_docs?: number + atc_docs?: number error?: string url?: string } +type KbHealth = { ok?: boolean; level?: string; ui_url?: string; url?: string; version?: string; error?: string } +type IndexInfo = { name: string; docs: number; size_bytes: number; health?: string; atc?: boolean } +type Field = { name: string; type?: string; aggregatable?: boolean } +type Hit = { index?: string; id?: string; score?: number; source?: Record } +type ReindexState = { + running?: boolean + current?: string | null + total_docs?: number + index_count?: number + errors?: string[] + log?: string[] + finished_at?: string | null +} +type Filt = { field: string; value: string } -type KbHealth = { ok?: boolean; level?: string; ui_url?: string; url?: string; error?: string } +const SOURCE_COLORS: Record = { + iceberg: '#34d399', + postgres_sales: '#38bdf8', + mysql_hr: '#f59e0b', + mongodb_supplychain: '#22c55e', + cassandra_telemetry: '#a78bfa', + neo4j: '#f472b6', + catalog: '#94a3b8', +} + +function fmtBytes(n: number) { + if (!n) return '0 B' + const u = ['B', 'KB', 'MB', 'GB', 'TB'] + let i = 0 + let v = n + while (v >= 1024 && i < u.length - 1) { v /= 1024; i++ } + return `${v.toFixed(i ? 1 : 0)} ${u[i]}` +} +function fmtVal(v: unknown): string { + if (v === null || v === undefined) return '—' + if (typeof v === 'object') return JSON.stringify(v) + return String(v) +} +function srcOf(index?: string): string { + if (!index) return 'other' + const m = index.replace(/^atc-/, '') + for (const k of Object.keys(SOURCE_COLORS)) { + if (m.startsWith(k.replace(/_/g, '-'))) return k + } + return m.split('-')[0] +} export function SearchView() { const [es, setEs] = useState(null) const [kb, setKb] = useState(null) const [loading, setLoading] = useState(false) - const [query, setQuery] = useState('') - const [hits, setHits] = useState<{ index?: string; score?: number; source?: Record }[]>([]) + const [indices, setIndices] = useState([]) + const [includeSystem, setIncludeSystem] = useState(false) + const [selected, setSelected] = useState>(new Set()) + + const [q, setQuery] = useState('') + const [filters, setFilters] = useState([]) + const [size, setSize] = useState(20) + const [from, setFrom] = useState(0) + const [sortField, setSortField] = useState('') + const [results, setResults] = useState<{ total: number; took?: number; hits: Hit[] } | null>(null) + const [searching, setSearching] = useState(false) const [searchError, setSearchError] = useState(null) + const [expanded, setExpanded] = useState(null) + const [showRaw, setShowRaw] = useState(null) + + const [sourceAgg, setSourceAgg] = useState<{ key: string; count: number }[]>([]) + const [reindex, setReindex] = useState(null) + const [kbMsg, setKbMsg] = useState(null) + const pollRef = useRef | null>(null) + + // facets + const [facetIndex, setFacetIndex] = useState('') + const [facetFields, setFacetFields] = useState([]) + const [facetField, setFacetField] = useState('') + const [facetBuckets, setFacetBuckets] = useState<{ key: string; count: number }[]>([]) const load = useCallback(async () => { setLoading(true) @@ -37,117 +107,381 @@ export function SearchView() { } }, []) - useEffect(() => { - load() - const iv = setInterval(load, 15000) - return () => clearInterval(iv) - }, [load]) - - const onSearch = async () => { - if (!query.trim()) return - setSearchError(null) + const loadIndices = useCallback(async () => { try { - const r = await fetch(`/api/search/elasticsearch/query?q=${encodeURIComponent(query)}&size=15`) + const r = await fetch(`/api/search/indices?include_system=${includeSystem}`) + const j = await r.json() + if (j.ok) setIndices(j.indices || []) + } catch { /* ignore */ } + }, [includeSystem]) + + const loadSourceAgg = useCallback(async () => { + try { + const r = await fetch('/api/search/aggs?index=atc-*&field=meta.catalog.keyword&size=20') + const j = await r.json() + if (j.ok) setSourceAgg(j.buckets || []) + } catch { /* ignore */ } + }, []) + + useEffect(() => { load(); loadSourceAgg() }, [load, loadSourceAgg]) + useEffect(() => { loadIndices() }, [loadIndices]) + + const runSearch = useCallback(async (resetFrom = true) => { + setSearching(true) + setSearchError(null) + const nextFrom = resetFrom ? 0 : from + if (resetFrom) setFrom(0) + try { + const r = await fetch('/api/search/query', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ + q, + indices: [...selected], + filters, + from: nextFrom, + size, + sort: sortField ? { field: sortField, order: 'desc' } : undefined, + }), + }) const j = await r.json() if (!r.ok || !j.ok) { - setSearchError(j.error || 'Search failed — set ELASTIC_PASSWORD in .env for query API') - setHits([]) + setSearchError(j.error || 'Search failed') + setResults(null) return } - setHits(j.hits || []) + setResults({ total: j.total, took: j.took, hits: j.hits || [] }) } catch { setSearchError('Search request failed') + } finally { + setSearching(false) + } + }, [q, selected, filters, from, size, sortField]) + + // re-run when paging/size/filters change + useEffect(() => { runSearch(false) /* eslint-disable-next-line */ }, [from]) + useEffect(() => { if (results) runSearch(true) /* eslint-disable-next-line */ }, [filters, size]) + + const addFilter = (field: string, value: string) => { + setFilters((f) => (f.some((x) => x.field === field && x.value === value) ? f : [...f, { field, value }])) + } + + const toggleIndex = (name: string) => { + setSelected((s) => { + const n = new Set(s) + if (n.has(name)) n.delete(name) + else n.add(name) + return n + }) + } + + const startReindex = async () => { + setKbMsg(null) + await fetch('/api/search/reindex', { method: 'POST' }).catch(() => {}) + if (pollRef.current) clearInterval(pollRef.current) + pollRef.current = setInterval(async () => { + try { + const r = await fetch('/api/search/reindex/status') + const j = await r.json() + setReindex(j) + if (!j.running) { + if (pollRef.current) clearInterval(pollRef.current) + load(); loadIndices(); loadSourceAgg() + } + } catch { /* ignore */ } + }, 1500) + } + useEffect(() => () => { if (pollRef.current) clearInterval(pollRef.current) }, []) + + const setupKibana = async () => { + setKbMsg('Setting up Kibana…') + try { + const r = await fetch('/api/search/kibana/setup', { method: 'POST' }) + const j = await r.json() + setKbMsg(j.ok ? `Kibana ready — created ${j.created?.length ?? 0} objects (dashboard + data views).` : `Kibana setup failed: ${j.error}`) + } catch { + setKbMsg('Kibana setup request failed') } } + const loadFacetFields = async (index: string) => { + setFacetIndex(index) + setFacetField('') + setFacetBuckets([]) + setFacetFields([]) + if (!index) return + try { + const r = await fetch(`/api/search/mapping?index=${encodeURIComponent(index)}`) + const j = await r.json() + if (j.ok) setFacetFields((j.fields || []).filter((f: Field) => f.aggregatable)) + } catch { /* ignore */ } + } + const loadFacet = async (field: string) => { + setFacetField(field) + if (!facetIndex || !field) return + try { + const r = await fetch(`/api/search/aggs?index=${encodeURIComponent(facetIndex)}&field=${encodeURIComponent(field)}&size=15`) + const j = await r.json() + if (j.ok) setFacetBuckets(j.buckets || []) + } catch { /* ignore */ } + } + const healthColor = (h?: string) => (h === 'green' ? 'text-success' : h === 'yellow' ? 'text-warning' : 'text-danger') + const kibanaBase = kb?.ui_url || '' + + const topIndices = useMemo(() => [...indices].filter((i) => i.atc).sort((a, b) => b.docs - a.docs).slice(0, 10), [indices]) + const maxDocs = Math.max(1, ...topIndices.map((i) => i.docs)) + const aggMax = Math.max(1, ...sourceAgg.map((b) => b.count)) return (
+ {/* header */}

- - Elasticsearch & Kibana + Elasticsearch & Kibana + + {es?.ok ? (es.health || 'up') : 'offline'} +

-

atc-elastic01 · 10.0.21.46 · login: admin or elastic

+

+ {es?.cluster_name || 'atc-lakehouse'} · {es?.version ? `v${es.version}` : '10.0.21.46'} · full-text search across every indexed source +

-
- {kb?.ui_url && ( - - Kibana - +
+ {kibanaBase && ( + <> + + Dashboard + + + Kibana + + )} - + +
-
- - + {/* reindex / kibana banner */} + {(reindex?.running || kbMsg) && ( +
+ {reindex?.running ? ( + + + Indexing {reindex.current} · {reindex.total_docs} docs · {reindex.index_count} indices + + ) : ( + {kbMsg}{reindex?.finished_at && ` · last index run done (${reindex.total_docs} docs, ${reindex.index_count} indices${reindex.errors?.length ? `, ${reindex.errors.length} skipped` : ''})`} + )} +
+ )} + + {/* KPI row */} +
+ + - + + +
-
- setQuery(e.target.value)} - onKeyDown={(e) => e.key === 'Enter' && onSearch()} - placeholder="Search indices…" - className="min-w-0 flex-1 rounded border border-border bg-surface-overlay px-3 py-2 text-[12px]" - /> - + {/* charts */} +
+
+

Documents by source

+
+ {sourceAgg.map((b) => ( +
+ {b.key} +
+
+
+ {b.count} +
+ ))} + {sourceAgg.length === 0 &&

No indexed data yet — click “Re-index all”.

} +
+
+
+

Top indices

+
+ {topIndices.map((i) => ( +
+ {i.name.replace(/^atc-/, '')} +
+
+
+ {i.docs} +
+ ))} +
+
+
+ + {/* search bar */} +
+
+ setQuery(e.target.value)} + onKeyDown={(e) => e.key === 'Enter' && runSearch(true)} + placeholder="Search all data… (e.g. region:EMEA OR status:shipped OR free text)" + className="min-w-0 flex-1 rounded border border-border bg-surface-overlay px-3 py-2 text-[12px]" + /> + + +
+ {(filters.length > 0 || selected.size > 0) && ( +
+ {selected.size > 0 && ( + + {selected.size} {selected.size === 1 ? 'index' : 'indices'} scoped + + + )} + {filters.map((f, i) => ( + + {f.field}: {f.value} + + + ))} +
+ )}
{searchError &&

{searchError}

} -
-
-

Indices

- {es?.indices && es.indices.length > 0 ? ( - - - - - - - - - - - {es.indices.map((i) => ( - - - - - - - ))} - -
IndexDocsSizeHealth
{i.name}{i.docs ?? '—'}{i.size ?? '—'}{i.health}
- ) : ( -

- {es?.ok ? 'No indices listed.' : (es?.error || 'Elasticsearch :9200 not reachable from Command Center — Kibana :5601 may still be up.')} -

- )} + {/* body: indices | results | facets */} +
+ {/* indices */} +
+
+ Indices ({indices.length}) + +
+
+ {indices.map((i) => ( +
+ toggleIndex(i.name)} className="shrink-0" /> + + +
+ ))} +
- {hits.length > 0 && ( -
-

Search results

+ {/* results */} +
+
+ {results ? `${results.total.toLocaleString()} hits · ${results.took ?? 0} ms` : 'Run a search to see results'} + {results && results.total > size && ( + + + {from + 1}–{Math.min(from + size, results.total)} + + + )} +
+
+ {!results &&

Search across {(es?.atc_docs ?? 0).toLocaleString()} indexed records from Postgres, MySQL, MongoDB, Cassandra, Neo4j and the Iceberg lakehouse.

} + {results?.hits.length === 0 &&

No matches.

}
- {hits.map((h, i) => ( -
-

{h.index} · score {h.score?.toFixed(2)}

- 
{JSON.stringify(h.source, null, 2).slice(0, 400)}
-
- ))} + {results?.hits.map((h) => { + const key = `${h.index}-${h.id}` + const src = (h.source || {}) as Record + const meta = (src.meta || {}) as Record + const fields = Object.entries(src).filter(([k]) => k !== 'meta') + const isOpen = expanded === key + return ( +
+ + {isOpen && ( +
+
+ id: {h.id} + {meta.fqn ? · source: {String(meta.fqn)} : null} + +
+ {showRaw === key ? ( + 
{JSON.stringify(src, null, 2)}
+ ) : ( + + + {fields.map(([k, v]) => ( + + + + + ))} + +
{k} + {fmtVal(v)} + +
+ )} +
+ )} +
+ ) + })}
- )} +
+ + {/* facets */} +
+
+ Facets +
+
+ {!facetIndex &&

Click an index name on the left to explore its fields and top values.

} + {facetIndex && ( + <> +

{facetIndex.replace(/^atc-/, '')}

+ +
+ {facetBuckets.map((b) => ( + + ))} + {facetField && facetBuckets.length === 0 &&

No aggregatable values.

} +
+ + )} +
+
) @@ -157,7 +491,7 @@ function Stat({ label, value, className }: { label: string; value: string; class return (

{label}

-

{value}

+

{value}

) } diff --git a/ui/src/components/layout/SideNav.tsx b/ui/src/components/layout/SideNav.tsx index 76bba29..2881e15 100644 --- a/ui/src/components/layout/SideNav.tsx +++ b/ui/src/components/layout/SideNav.tsx @@ -1,4 +1,5 @@ -import { Database, DatabaseZap, HardDrive, Search, LayoutDashboard, MessageSquare, Server, TerminalSquare, Activity, GitBranch } from 'lucide-react' +import { useEffect, useState } from 'react' +import { Database, DatabaseZap, HardDrive, Search, LayoutDashboard, MessageSquare, Server, TerminalSquare, Activity, GitBranch, ExternalLink } from 'lucide-react' import type { GpuStatus, WorkloadData } from '../../types' import type { GpuLiveMetrics } from '../../hooks/useLiveMetrics' import { cn } from '../../lib/utils' @@ -47,6 +48,13 @@ export function SideNav({ onOpenSsh, }: Props) { const matrixBoost = gpuBoost || mainView === 'knowledge' + const [kibanaUrl, setKibanaUrl] = useState(null) + useEffect(() => { + fetch('/api/search/health') + .then((r) => (r.ok ? r.json() : null)) + .then((j) => { if (j?.kibana?.ui_url) setKibanaUrl(j.kibana.ui_url) }) + .catch(() => {}) + }, []) return (