diff --git a/api/Dockerfile b/api/Dockerfile index 0f4a57a..0cf56f4 100644 --- a/api/Dockerfile +++ b/api/Dockerfile @@ -4,7 +4,7 @@ WORKDIR /app RUN apt-get update && apt-get install -y --no-install-recommends curl && rm -rf /var/lib/apt/lists/* COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt -COPY main.py lab_context.py agent_terminal.py workload.py node_registry.py node_ops.py topology_views.py supervisor.py approval_service.py db.py dockhand_envs.py presentation.py database_inventory.py presentation_upload.py presentation_static.py storage_s3.py . +COPY main.py lab_context.py agent_terminal.py workload.py node_registry.py node_ops.py topology_views.py supervisor.py approval_service.py db.py dockhand_envs.py presentation.py database_inventory.py presentation_upload.py presentation_static.py storage_s3.py elasticsearch_api.py . RUN mkdir -p /data ENV DATABASE_URL=sqlite:////data/atc-agents.db EXPOSE 3201 diff --git a/api/elasticsearch_api.py b/api/elasticsearch_api.py new file mode 100644 index 0000000..9ac08e5 --- /dev/null +++ b/api/elasticsearch_api.py @@ -0,0 +1,120 @@ +"""Elasticsearch + Kibana health API for Command Center.""" + +from __future__ import annotations + +import os +from typing import Any + +import httpx +from fastapi import APIRouter, Query +from fastapi.responses import JSONResponse + +ELASTICSEARCH_URL = os.getenv("ELASTICSEARCH_URL", "https://10.0.21.46:9200").rstrip("/") +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", "") + +router = APIRouter(prefix="/api/search", tags=["search"]) + + +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) + 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 = 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) + indices = stats_r.json() if stats_r.status_code < 400 else [] + return { + "ok": True, + "url": ELASTICSEARCH_URL, + "cluster_name": info.get("cluster_name"), + "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": [ + { + "name": i.get("index"), + "docs": i.get("docs.count"), + "size": i.get("store.size"), + "health": i.get("health"), + } + for i in (indices[:50] if isinstance(indices, list) else []) + ], + } + except Exception as exc: + return {"ok": False, "url": ELASTICSEARCH_URL, "error": str(exc)} + + +async def _probe_kibana() -> dict[str, Any]: + try: + 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} + data = r.json() + overall = data.get("status", {}).get("overall", {}) + return { + "ok": overall.get("level") in ("available", "green", "yellow"), + "url": KIBANA_URL, + "ui_url": KIBANA_URL, + "level": overall.get("level", "unknown"), + } + except Exception as exc: + return {"ok": False, "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, + } + + +@router.get("/elasticsearch") +async def get_elasticsearch(): + result = await _probe_es() + if not result.get("ok"): + return JSONResponse(result, status_code=502) + return result + + +@router.get("/kibana") +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) + 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}, + ) + 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: + return JSONResponse({"ok": False, "error": str(exc)}, status_code=502) diff --git a/api/main.py b/api/main.py index 6cce657..677fe2c 100644 --- a/api/main.py +++ b/api/main.py @@ -28,6 +28,7 @@ from presentation import build_presentation_payload, render_presentation_html from presentation_upload import get_deck, list_decks, save_upload from presentation_static import get_static_deck, list_static_decks from storage_s3 import router as storage_s3_router +from elasticsearch_api import router as elasticsearch_router from node_registry import NODE_IDS, NODE_AGENT, NODE_REGISTRY, is_node_id from node_ops import build_node_detail, probe_node, run_node_probe_task from approval_service import ( @@ -705,6 +706,7 @@ async def lifespan(app: FastAPI): app = FastAPI(title="ATC Command Center API", lifespan=lifespan) app.include_router(storage_s3_router) +app.include_router(elasticsearch_router) app.add_middleware( CORSMiddleware, allow_origins=["*"], diff --git a/api/node_registry.py b/api/node_registry.py index b898c5f..d0c33de 100644 --- a/api/node_registry.py +++ b/api/node_registry.py @@ -160,6 +160,26 @@ NODE_REGISTRY: dict[str, dict[str, Any]] = { ], "commands": ["hdfs dfsadmin -report", "datanode status", "block health"], }, + + "elastic": { + "label": "Elasticsearch / Kibana", + "vm": "atc-elastic01", + "vmid": 0, + "pve": "atc-gpu", + "ip": "10.0.21.46", + "role": "search", + "color": "#f5a623", + "description": "Elasticsearch cluster atc-lakehouse + Kibana dashboards on port 5601.", + "links": [ + {"label": "Kibana", "url": "http://10.0.21.46:5601"}, + {"label": "Elasticsearch", "url": "https://10.0.21.46:9200"}, + ], + "endpoints": [ + {"name": "elasticsearch", "host": "10.0.21.46", "port": "9200", "proto": "https"}, + {"name": "kibana", "host": "10.0.21.46", "port": "5601", "proto": "http"}, + ], + "commands": ["cluster health", "cat indices", "kibana status"], + }, "gpu": { "label": "GPU Lab", "vm": "atc-gpu-dev", diff --git a/api/presentation_static.py b/api/presentation_static.py index dad3c52..edc9630 100644 --- a/api/presentation_static.py +++ b/api/presentation_static.py @@ -108,6 +108,7 @@ MODERN_DECKS: dict[str, dict[str, Any]] = { "React UI — Data Platform, Presentation, Data Quality, Knowledge Chat", "Docling on port 5001 for document parsing UI + API", "Postgres + Redis for agents; ChromaDB for vectors", + "Elasticsearch + Kibana on atc-elastic01 (10.0.21.46)", ], }, { @@ -161,7 +162,7 @@ MODERN_DECKS: dict[str, dict[str, Any]] = { "bullets": [ "Command Center VM304: 10.0.21.33 (this dashboard)", "GPU Lab VM303: 10.0.20.106 — 7× V100, vLLM, model manager", - "DB Vault: 10.0.21.51 · Lakehouse: 10.0.21.50", + "DB Vault: 10.0.21.51 · Lakehouse: 10.0.21.50 · Elastic: 10.0.21.46", "Docling UI: http://10.0.21.33:5001/ui/", ], }, diff --git a/config/command-center/.env.example b/config/command-center/.env.example index abd5018..43e5b7a 100644 --- a/config/command-center/.env.example +++ b/config/command-center/.env.example @@ -1,27 +1,9 @@ -# Command Center — VM304 (10.0.21.33) -# Copy to /opt/atc-agents/.env — never commit secrets - -# Postgres (internal) -POSTGRES_USER=atc -POSTGRES_PASSWORD=change-me -POSTGRES_DB=atc_agents - -# GPU / LLM (VM303) -GPU_URL=http://10.0.20.106:9000 -LLM_URL=http://10.0.20.106:8001/v1 -LLM_MODEL=gpt-4o -LLM_API_KEY=sk-local - -# ObjectScale S3 (VM objectscale 10.0.20.111) S3_ENDPOINT=http://10.0.20.111:9020 -S3_ACCESS_KEY=object_admin1 -S3_SECRET_KEY=REDACTED-use-deploy-yml-or-ecs-admin +S3_ACCESS_KEY=your-access-key-id +S3_SECRET_KEY=your-secret-key S3_REGION=us-east-1 - -# Jupyter -JUPYTER_TOKEN=change-me-jupyter-token - -# Lakehouse / ETL (optional probes) -LAKEHOUSE_HOST=10.0.21.50 -AIRFLOW_URL=http://10.0.21.55:8080 -KAFKA_UI_URL=http://10.0.21.36:9000 +JUPYTER_TOKEN=change-me +ELASTICSEARCH_URL=https://10.0.21.46:9200 +KIBANA_URL=http://10.0.21.46:5601 +ELASTIC_USER=elastic +ELASTIC_PASSWORD=change-me diff --git a/config/command-center/docker-compose.yml b/config/command-center/docker-compose.yml index 70a0b57..e1cc2f7 100644 --- a/config/command-center/docker-compose.yml +++ b/config/command-center/docker-compose.yml @@ -45,6 +45,10 @@ services: S3_ACCESS_KEY: ${S3_ACCESS_KEY:-object_admin1} S3_SECRET_KEY: ${S3_SECRET_KEY} S3_REGION: ${S3_REGION:-us-east-1} + ELASTICSEARCH_URL: ${ELASTICSEARCH_URL:-https://10.0.21.46:9200} + KIBANA_URL: ${KIBANA_URL:-http://10.0.21.46:5601} + ELASTIC_USER: ${ELASTIC_USER:-elastic} + ELASTIC_PASSWORD: ${ELASTIC_PASSWORD:-} volumes: - api_data:/data depends_on: @@ -116,6 +120,8 @@ services: AWS_SECRET_ACCESS_KEY: ${S3_SECRET_KEY} S3_ENDPOINT: ${S3_ENDPOINT:-http://10.0.20.111:9020} AWS_DEFAULT_REGION: ${S3_REGION:-us-east-1} + ELASTICSEARCH_URL: ${ELASTICSEARCH_URL:-https://10.0.21.46:9200} + KIBANA_URL: ${KIBANA_URL:-http://10.0.21.46:5601} command: > start-notebook.sh --NotebookApp.base_url=/jupyter/ diff --git a/docker-compose.yml b/docker-compose.yml index 70a0b57..e1cc2f7 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -45,6 +45,10 @@ services: S3_ACCESS_KEY: ${S3_ACCESS_KEY:-object_admin1} S3_SECRET_KEY: ${S3_SECRET_KEY} S3_REGION: ${S3_REGION:-us-east-1} + ELASTICSEARCH_URL: ${ELASTICSEARCH_URL:-https://10.0.21.46:9200} + KIBANA_URL: ${KIBANA_URL:-http://10.0.21.46:5601} + ELASTIC_USER: ${ELASTIC_USER:-elastic} + ELASTIC_PASSWORD: ${ELASTIC_PASSWORD:-} volumes: - api_data:/data depends_on: @@ -116,6 +120,8 @@ services: AWS_SECRET_ACCESS_KEY: ${S3_SECRET_KEY} S3_ENDPOINT: ${S3_ENDPOINT:-http://10.0.20.111:9020} AWS_DEFAULT_REGION: ${S3_REGION:-us-east-1} + ELASTICSEARCH_URL: ${ELASTICSEARCH_URL:-https://10.0.21.46:9200} + KIBANA_URL: ${KIBANA_URL:-http://10.0.21.46:5601} command: > start-notebook.sh --NotebookApp.base_url=/jupyter/ diff --git a/ui/src/App.tsx b/ui/src/App.tsx index abbd7ae..9e87d0c 100644 --- a/ui/src/App.tsx +++ b/ui/src/App.tsx @@ -15,6 +15,7 @@ import { PresentationView } from './components/features/PresentationView' import { DataQualityView } from './components/features/DataQualityView' import { KnowledgeChatView } from './components/features/KnowledgeChatView' import { StorageView } from './components/features/StorageView' +import { SearchView } from './components/features/SearchView' import { TerminalDock } from './components/features/TerminalDock' import { resolveInfraNode } from './lib/infraCatalog' import { cn } from './lib/utils' @@ -118,6 +119,8 @@ export default function App() { ) : cc.mainView === 'storage' ? ( + ) : cc.mainView === 'search' ? ( + ) : ( )} diff --git a/ui/src/components/features/PlatformTopology.tsx b/ui/src/components/features/PlatformTopology.tsx index 055b1c8..a967816 100644 --- a/ui/src/components/features/PlatformTopology.tsx +++ b/ui/src/components/features/PlatformTopology.tsx @@ -63,6 +63,8 @@ const STAGES: TopoStage[] = [ nodes: [ { id: 'bi', label: 'BI / Reporting', sub: 'Dashboards', metricKey: 'bi' }, { id: 'jupyter', label: 'Jupyter Notebooks', sub: 'Data science', metricKey: 'jupyter' }, + { id: 'elasticsearch', label: 'Elasticsearch', sub: 'Search · :9200', metricKey: 'elasticsearch' }, + { id: 'kibana', label: 'Kibana', sub: 'Dashboards · :5601', metricKey: 'kibana' }, { id: 'llm', label: 'GenAI LLM', sub: 'vLLM inference', metricKey: 'llm' }, ], }, @@ -95,6 +97,10 @@ const FLOW_EDGES: FlowEdge[] = [ { from: 's3', to: 'jupyter', kind: 'serve', label: 'Datasets' }, { from: 'trino', to: 'llm', kind: 'serve', label: 'RAG context' }, { from: 's3', to: 'llm', kind: 'serve', label: 'Model artifacts' }, + { from: 'kafka', to: 'elasticsearch', kind: 'stream', label: 'Index' }, + { from: 'spark', to: 'elasticsearch', kind: 'etl', label: 'Bulk index' }, + { from: 'elasticsearch', to: 'kibana', kind: 'serve', label: 'Visualize' }, + { from: 'elasticsearch', to: 'bi', kind: 'serve', label: 'Search' }, ] const STAGE_BADGE: Record = { @@ -136,7 +142,7 @@ const NODE_CLICK_MAP: Record = { postgresql: 'src-postgres', mysql: 'src-mysql', mongodb: 'src-mongo', cassandra: 'src-cassandra', debezium: 'cdc-postgres', kafka: 'stream-kafka', airflow: 'src-airflow', spark: 'stream-spark', trino: 'query-trino', iceberg: 'lake-iceberg', s3: 'lake-s3', bi: 'cons-bi', - jupyter: 'cons-notebooks', llm: 'cons-ml', + jupyter: 'cons-notebooks', elasticsearch: 'cons-elastic', kibana: 'cons-kibana', llm: 'cons-ml', } const NODE_POS: Record = {} @@ -175,7 +181,7 @@ function seedMetrics(): MetricState { postgresql: '12.4k rows/s', mysql: '8.1k rows/s', mongodb: '2.3k docs/s', cassandra: '5.6k ops/s', debezium: '4 connectors active', kafka: '142 MB/s', airflow: '18 DAGs · daily 02:00 UTC', spark: '6 executors live', trino: '3 queries active', iceberg: '847 tables · 2.1 TB', s3: '14.2 TB stored', - bi: '26 dashboards', jupyter: '12 kernels active', llm: 'Checking…', + bi: '26 dashboards', jupyter: '12 kernels active', elasticsearch: 'atc-lakehouse', kibana: 'available', llm: 'Checking…', } } @@ -239,6 +245,18 @@ export function PlatformTopology({ workload, animations, selectedNodeId, onNodeC const edgesLive = pipelineActive || anyBusy + useEffect(() => { + fetch("/api/search/health").then(r => r.json()).then(j => { + const kb = j.kibana + const es = j.elasticsearch + setMetrics(prev => ({ + ...prev, + kibana: kb?.ok ? (kb.level || "available") : "offline", + elasticsearch: es?.ok ? (es.health || "green") : (kb?.ok ? "via Kibana" : "offline"), + })) + }).catch(() => {}) + }, []) + useEffect(() => { setMetrics((prev) => ({ ...prev, llm: formatLlmMetric(workload) })) }, [workload?.gpu?.model, workload?.gpu?.inference_active, workload?.gpu?.avg_util, workload?.gpu?.gpus]) diff --git a/ui/src/components/features/SearchView.tsx b/ui/src/components/features/SearchView.tsx new file mode 100644 index 0000000..8c5dd32 --- /dev/null +++ b/ui/src/components/features/SearchView.tsx @@ -0,0 +1,163 @@ +import { useCallback, useEffect, useState } from 'react' +import { ExternalLink, Loader2, RefreshCw, Search } from 'lucide-react' +import { cn } from '../../lib/utils' +import { subTabActive, subTabIdle } from '../../lib/tabActive' + +type EsHealth = { + ok?: boolean + cluster_name?: string + version?: string + health?: string + nodes?: number + indices_count?: number + indices?: { name?: string; docs?: string; size?: string; health?: string }[] + error?: string + url?: string +} + +type KbHealth = { ok?: boolean; level?: string; ui_url?: string; url?: string; error?: string } + +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 [searchError, setSearchError] = useState(null) + + const load = useCallback(async () => { + setLoading(true) + try { + const r = await fetch('/api/search/health') + const j = await r.json() + setEs(j.elasticsearch || null) + setKb(j.kibana || null) + } finally { + setLoading(false) + } + }, []) + + useEffect(() => { + load() + const iv = setInterval(load, 15000) + return () => clearInterval(iv) + }, [load]) + + const onSearch = async () => { + if (!query.trim()) return + setSearchError(null) + try { + const r = await fetch(`/api/search/elasticsearch/query?q=${encodeURIComponent(query)}&size=15`) + const j = await r.json() + if (!r.ok || !j.ok) { + setSearchError(j.error || 'Search failed — set ELASTIC_PASSWORD in .env for query API') + setHits([]) + return + } + setHits(j.hits || []) + } catch { + setSearchError('Search request failed') + } + } + + const healthColor = (h?: string) => (h === 'green' ? 'text-success' : h === 'yellow' ? 'text-warning' : 'text-danger') + + return ( +
+
+
+

+ + Elasticsearch & Kibana +

+

atc-elastic01 · 10.0.21.46 · cluster atc-lakehouse

+
+
+ {kb?.ui_url && ( + + Kibana + + )} + +
+
+ +
+ + + + +
+ +
+ setQuery(e.target.value)} + onKeyDown={(e) => e.key === 'Enter' && onSearch()} + placeholder="Search indices (requires ELASTIC_PASSWORD)…" + className="min-w-0 flex-1 rounded border border-border bg-surface-overlay px-3 py-2 text-[12px]" + /> + +
+ + {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.')} +

+ )} +
+ + {hits.length > 0 && ( +
+

Search results

+
+ {hits.map((h, i) => ( +
+

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

+ 
{JSON.stringify(h.source, null, 2).slice(0, 400)}
+
+ ))} +
+
+ )} +
+
+ ) +} + +function Stat({ label, value, className }: { label: string; value: string; className?: string }) { + return ( +
+

{label}

+

{value}

+
+ ) +} diff --git a/ui/src/components/layout/SideNav.tsx b/ui/src/components/layout/SideNav.tsx index 85c710f..7ef88a0 100644 --- a/ui/src/components/layout/SideNav.tsx +++ b/ui/src/components/layout/SideNav.tsx @@ -1,4 +1,4 @@ -import { DatabaseZap, HardDrive, LayoutDashboard, MessageSquare, Presentation, ShieldCheck } from 'lucide-react' +import { DatabaseZap, HardDrive, Search, LayoutDashboard, MessageSquare, Presentation, ShieldCheck } from 'lucide-react' import type { Agent, AgentAnim, GpuStatus } from '../../types' import type { GpuLiveMetrics } from '../../hooks/useLiveMetrics' import { getAgentMeta } from '../../lib/agentMeta' @@ -6,7 +6,7 @@ import { cn } from '../../lib/utils' import { viewTabActive, viewTabIdle } from '../../lib/tabActive' import { GpuMatrixPanel } from '../features/GpuMatrixPanel' -type MainView = 'platform' | 'presentation' | 'dataquality' | 'knowledge' | 'storage' | 'approvals' +type MainView = 'platform' | 'presentation' | 'dataquality' | 'knowledge' | 'storage' | 'search' | 'approvals' type Props = { agents: Agent[] @@ -31,6 +31,7 @@ const VIEWS: { id: MainView; label: string; icon: typeof LayoutDashboard }[] = [ { id: 'dataquality', label: 'Data Quality', icon: DatabaseZap }, { id: 'knowledge', label: 'Knowledge Chat', icon: MessageSquare }, { id: 'storage', label: 'Object Storage', icon: HardDrive }, + { id: 'search', label: 'Elasticsearch', icon: Search }, ] export function SideNav({ diff --git a/ui/src/hooks/useCommandCenter.ts b/ui/src/hooks/useCommandCenter.ts index 53d3bda..30150b3 100644 --- a/ui/src/hooks/useCommandCenter.ts +++ b/ui/src/hooks/useCommandCenter.ts @@ -52,7 +52,7 @@ export function useCommandCenter() { const [selectedNode, setSelectedNode] = useState(null) const [nodeDetail, setNodeDetail] = useState(null) const [nodeBusy, setNodeBusy] = useState(false) - const [mainView, setMainView] = useState<'platform' | 'approvals' | 'presentation' | 'dataquality' | 'knowledge' | 'storage'>('platform') + const [mainView, setMainView] = useState<'platform' | 'approvals' | 'presentation' | 'dataquality' | 'knowledge' | 'storage' | 'search'>('platform') const [approvalHighlight, setApprovalHighlight] = useState(false) const [chatExpanded, setChatExpanded] = useState(false) const promptTimeoutRef = useRef | null>(null) diff --git a/ui/src/lib/constants.ts b/ui/src/lib/constants.ts index 4ddfb62..e41912c 100644 --- a/ui/src/lib/constants.ts +++ b/ui/src/lib/constants.ts @@ -16,6 +16,8 @@ export const NODE_ALIASES: Record = { 'query-trino': 'lakehouse', 'query-dbt': 'lakehouse', 'cons-bi': 'docker', + 'cons-elastic': 'docker', + 'cons-kibana': 'docker', 'cons-notebooks': 'lakehouse', 'cons-ml': 'gpu', }