mirror of
https://github.com/hanzoai/bootnode.git
synced 2026-09-30 20:46:32 +00:00
feat(nodes): k8s detector — discover & adopt operator-managed validators (#15)
Bootnode could only see nodes it launched as Docker containers; real Lux/Zoo
validators run as K8s workloads behind operator CRDs and were invisible to
GET /v1/nodes. Add a detector that scans a cluster's kubeconfig and adopts
its running validators into the DB.
- core/k8s_nodes.py: K8sNodeDetector scans LuxNetwork (lux.network/v1alpha1
luxnetworks) + ZooChain (zoo.network/v1alpha1 zoochains) CRs across all
namespaces via CustomObjectsApi, enriches each with its luxd/zood pods for
live status, and upserts Cluster + Node rows. Adoption, not recreation:
existing rows (keyed on (cluster_id, pod_name)) are updated in place, so
re-scans are idempotent. New operator chains are one descriptor row.
- db/models.py: add Cluster + Node models (portable JSON/Uuid types, matching
the existing db_nodes.py / seed_nodes.py consumers that imported them).
- api/nodes: merge DB-adopted k8s nodes into list_nodes() alongside Docker
discovery; add POST /v1/nodes/detect and POST /v1/nodes/clusters/{id}/scan
(kubeconfig resolved via the shared fleets path). Reuses existing get_db.
- pyproject: add aiosqlite (config.database_url now defaults to
sqlite+aiosqlite; the driver was missing).
Detector adopt/idempotency + model mapping verified against in-memory SQLite
with faked LuxNetwork/ZooChain CRs and pods; full app OpenAPI builds with the
new routes; 112 existing tests pass (pre-existing test_lux_fleet /
test_chain_config failures unchanged).
This commit is contained in:
@@ -11,8 +11,18 @@ from datetime import datetime
|
||||
import docker
|
||||
import os
|
||||
import asyncio
|
||||
import logging
|
||||
import uuid
|
||||
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from bootnode.db.session import get_db
|
||||
from bootnode.db.models import Node as DBNode
|
||||
from bootnode.core.k8s_nodes import detector
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
# Node images by chain - execution and consensus clients
|
||||
@@ -973,12 +983,62 @@ async def start_node_container(node_id: str, node: dict):
|
||||
pass
|
||||
|
||||
|
||||
def _db_node_to_dict(n: "DBNode") -> dict:
|
||||
"""Map a persisted (K8s-adopted) Node row to the list_nodes() dict shape."""
|
||||
return {
|
||||
"id": str(n.id),
|
||||
"name": n.name,
|
||||
"chain": n.chain,
|
||||
"network": n.network,
|
||||
"provider": n.provider,
|
||||
"status": n.status,
|
||||
"endpoint": n.rpc_endpoint,
|
||||
"ws_endpoint": n.ws_endpoint,
|
||||
"container_id": n.container_id,
|
||||
"execution_client": n.execution_client,
|
||||
"consensus_client": n.consensus_client,
|
||||
"created_at": n.created_at.isoformat() if n.created_at else datetime.now().isoformat(),
|
||||
"last_health_check": (
|
||||
n.last_health_check.isoformat() if n.last_health_check else None
|
||||
),
|
||||
"metrics": {},
|
||||
}
|
||||
|
||||
|
||||
async def _db_discovered_nodes(db: AsyncSession) -> List[dict]:
|
||||
"""Load K8s-managed / adopted nodes persisted in the DB.
|
||||
|
||||
These are surfaced by the K8s detector (operator-managed validators) and
|
||||
are not visible to Docker — merging them here is what makes attaching a
|
||||
cluster's nodes appear in GET /v1/nodes alongside Docker-managed ones.
|
||||
"""
|
||||
try:
|
||||
result = await db.execute(
|
||||
select(DBNode).where(DBNode.provider == "kubernetes")
|
||||
)
|
||||
return [_db_node_to_dict(n) for n in result.scalars().all()]
|
||||
except Exception as e: # noqa: BLE001 - DB optional; never break Docker listing
|
||||
logger.debug("DB node discovery skipped: %s", e)
|
||||
return []
|
||||
|
||||
|
||||
@router.get("/", response_model=List[NodeResponse])
|
||||
async def list_nodes():
|
||||
"""List all managed blockchain nodes"""
|
||||
async def list_nodes(db: AsyncSession = Depends(get_db)):
|
||||
"""List all managed blockchain nodes.
|
||||
|
||||
Merges three sources: in-memory Docker state, live Docker containers, and
|
||||
DB-persisted K8s-managed validators (adopted via the detector).
|
||||
"""
|
||||
client = get_docker_client()
|
||||
nodes = list(_nodes.values())
|
||||
|
||||
# K8s-managed validators adopted into the DB (not visible to Docker).
|
||||
seen_db_ids = {str(n.get("id")) for n in nodes}
|
||||
for dbn in await _db_discovered_nodes(db):
|
||||
if dbn["id"] not in seen_db_ids:
|
||||
nodes.append(dbn)
|
||||
seen_db_ids.add(dbn["id"])
|
||||
|
||||
# Also check for any bootnode-managed containers not in memory
|
||||
if client:
|
||||
try:
|
||||
@@ -1026,6 +1086,84 @@ async def list_nodes():
|
||||
return nodes
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# K8s detection — discover & adopt operator-managed validators
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class DetectRequest(BaseModel):
|
||||
"""Scan a cluster for K8s-managed blockchain validators.
|
||||
|
||||
Supply either ``cluster_id`` (kubeconfig fetched via the infra/DO path) or
|
||||
a raw ``kubeconfig`` YAML string. With neither, the in-cluster service
|
||||
account is used.
|
||||
"""
|
||||
|
||||
cluster_id: Optional[str] = None
|
||||
kubeconfig: Optional[str] = None
|
||||
cluster_name: str = ""
|
||||
provider: str = "digitalocean"
|
||||
region: str = "sfo3"
|
||||
endpoint: Optional[str] = None
|
||||
|
||||
|
||||
async def _resolve_kubeconfig(
|
||||
cluster_id: Optional[str], kubeconfig: Optional[str]
|
||||
) -> Optional[str]:
|
||||
"""Resolve a kubeconfig YAML: explicit string wins, else fetch by cluster_id.
|
||||
|
||||
Reuses the fleets module's DO/in-cluster fetch so a cluster_id is turned
|
||||
into a kubeconfig exactly one way across the codebase.
|
||||
"""
|
||||
if kubeconfig:
|
||||
return kubeconfig
|
||||
if cluster_id:
|
||||
from bootnode.api.fleets import _get_kubeconfig
|
||||
|
||||
return await _get_kubeconfig(cluster_id)
|
||||
return None
|
||||
|
||||
|
||||
async def _run_detect(body: DetectRequest, db: AsyncSession) -> dict:
|
||||
cfg = await _resolve_kubeconfig(body.cluster_id, body.kubeconfig)
|
||||
result = await detector.scan_cluster(
|
||||
db,
|
||||
cluster_id=body.cluster_id or "incluster",
|
||||
kubeconfig=cfg,
|
||||
cluster_name=body.cluster_name,
|
||||
provider=body.provider,
|
||||
region=body.region,
|
||||
endpoint=body.endpoint,
|
||||
)
|
||||
return result.as_dict()
|
||||
|
||||
|
||||
@router.post("/detect")
|
||||
async def detect_nodes(
|
||||
body: DetectRequest,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
):
|
||||
"""Detect & adopt K8s-managed validators on a cluster.
|
||||
|
||||
Scans LuxNetwork + ZooChain CRs (and their luxd/zood pods), upserting each
|
||||
as a Node so it appears in ``GET /v1/nodes``. Idempotent: re-running adopts
|
||||
existing nodes rather than recreating them.
|
||||
"""
|
||||
return await _run_detect(body, db)
|
||||
|
||||
|
||||
@router.post("/clusters/{cluster_id}/scan")
|
||||
async def scan_cluster_nodes(
|
||||
cluster_id: str,
|
||||
body: Optional[DetectRequest] = None,
|
||||
db: AsyncSession = Depends(get_db),
|
||||
):
|
||||
"""Scan a specific cluster by id for K8s-managed validators (on demand)."""
|
||||
req = body or DetectRequest()
|
||||
req.cluster_id = cluster_id
|
||||
return await _run_detect(req, db)
|
||||
|
||||
|
||||
@router.get("/presets")
|
||||
async def list_presets():
|
||||
"""List available node presets for simple mode"""
|
||||
@@ -1345,13 +1483,13 @@ async def get_node_stats():
|
||||
|
||||
|
||||
@router.get("/metrics/all")
|
||||
async def get_all_node_metrics():
|
||||
async def get_all_node_metrics(db: AsyncSession = Depends(get_db)):
|
||||
"""Get metrics for all nodes"""
|
||||
import httpx
|
||||
nodes_metrics = {}
|
||||
|
||||
# First ensure we have discovered all containers
|
||||
await list_nodes()
|
||||
await list_nodes(db)
|
||||
|
||||
for node_id, node in _nodes.items():
|
||||
metrics = {}
|
||||
|
||||
@@ -0,0 +1,490 @@
|
||||
"""K8s node detector — discover & adopt cluster-managed blockchain validators.
|
||||
|
||||
Bootnode's Docker discovery (``api/nodes`` ``list_nodes``) can only see nodes it
|
||||
launched as containers. Real Lux / Zoo validators run as K8s workloads behind
|
||||
operator CRDs, invisible to Docker. This module closes that gap: given a
|
||||
cluster's kubeconfig it scans
|
||||
|
||||
- **LuxNetwork** CRs (group ``lux.network``, v1alpha1, plural ``luxnetworks``)
|
||||
- **ZooChain** CRs (group ``zoo.network``, v1alpha1, plural ``zoochains``)
|
||||
|
||||
across all namespaces, enriches each with its luxd / zood StatefulSet + pods for
|
||||
live status, and upserts the result as ``Cluster`` + ``Node`` rows so the
|
||||
validators surface in ``GET /v1/nodes``.
|
||||
|
||||
Adoption, not recreation: a node already in the DB (matched on its natural key
|
||||
``(cluster_id, pod_name)``) is updated in place. Re-scanning is idempotent.
|
||||
|
||||
The CRD shapes mirror ``core.chains.fleet_manager`` (single source of truth for
|
||||
LuxNetwork field names) — this detector reuses those constants and adds the Zoo
|
||||
equivalent, rather than re-deriving them.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import UTC, datetime
|
||||
from typing import Any, Optional
|
||||
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from bootnode.core.chains.fleet_manager import (
|
||||
LUX_CRD_GROUP,
|
||||
LUX_CRD_PLURAL,
|
||||
LUX_CRD_VERSION,
|
||||
)
|
||||
from bootnode.db.models import Cluster, Node
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Zoo operator CRD — mirrors the Lux constants in fleet_manager.
|
||||
ZOO_CRD_GROUP = "zoo.network"
|
||||
ZOO_CRD_VERSION = "v1alpha1"
|
||||
ZOO_CRD_PLURAL = "zoochains"
|
||||
|
||||
# Map an operator CR phase onto a Node.status value.
|
||||
_PHASE_TO_STATUS: dict[str, str] = {
|
||||
"Pending": "provisioning",
|
||||
"Creating": "provisioning",
|
||||
"Bootstrapping": "syncing",
|
||||
"Running": "running",
|
||||
"Degraded": "degraded",
|
||||
"Failed": "error",
|
||||
"Error": "error",
|
||||
}
|
||||
|
||||
# Per-chain CRD descriptor — one entry per operator we adopt from. Adding a new
|
||||
# operator-managed chain is one row here, nothing else.
|
||||
@dataclass(frozen=True)
|
||||
class _CRDSpec:
|
||||
chain: str
|
||||
group: str
|
||||
version: str
|
||||
plural: str
|
||||
execution_client: str
|
||||
image_default: str
|
||||
|
||||
|
||||
_CRDS: tuple[_CRDSpec, ...] = (
|
||||
_CRDSpec(
|
||||
chain="lux",
|
||||
group=LUX_CRD_GROUP,
|
||||
version=LUX_CRD_VERSION,
|
||||
plural=LUX_CRD_PLURAL,
|
||||
execution_client="luxd",
|
||||
image_default="ghcr.io/luxfi/node:latest",
|
||||
),
|
||||
_CRDSpec(
|
||||
chain="zoo",
|
||||
group=ZOO_CRD_GROUP,
|
||||
version=ZOO_CRD_VERSION,
|
||||
plural=ZOO_CRD_PLURAL,
|
||||
execution_client="zood",
|
||||
image_default="ghcr.io/zooai/node:latest",
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class ScanResult:
|
||||
"""Outcome of scanning one cluster."""
|
||||
|
||||
cluster_id: str
|
||||
cluster_db_id: Optional[str] = None
|
||||
crs_found: int = 0
|
||||
nodes_adopted: int = 0 # existing rows updated
|
||||
nodes_created: int = 0 # new rows inserted
|
||||
pods_seen: int = 0
|
||||
errors: list[str] = field(default_factory=list)
|
||||
|
||||
def as_dict(self) -> dict[str, Any]:
|
||||
return {
|
||||
"cluster_id": self.cluster_id,
|
||||
"cluster_db_id": self.cluster_db_id,
|
||||
"crs_found": self.crs_found,
|
||||
"nodes_adopted": self.nodes_adopted,
|
||||
"nodes_created": self.nodes_created,
|
||||
"nodes_total": self.nodes_adopted + self.nodes_created,
|
||||
"pods_seen": self.pods_seen,
|
||||
"errors": self.errors,
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# K8s client
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _k8s_apis(kubeconfig: Optional[str]):
|
||||
"""Return ``(CustomObjectsApi, CoreV1Api, AppsV1Api)`` for a kubeconfig.
|
||||
|
||||
Loads from the YAML string when given, else falls back to in-cluster
|
||||
config. Same loading shape as ``FleetManager._get_k8s_clients`` (kept
|
||||
local so the detector has no hidden coupling to that class).
|
||||
"""
|
||||
import os
|
||||
import tempfile
|
||||
|
||||
from kubernetes import client, config
|
||||
|
||||
if kubeconfig:
|
||||
tmp = tempfile.NamedTemporaryFile(
|
||||
mode="w", suffix=".kubeconfig", delete=False
|
||||
)
|
||||
try:
|
||||
tmp.write(kubeconfig)
|
||||
tmp.flush()
|
||||
tmp.close()
|
||||
config.load_kube_config(config_file=tmp.name)
|
||||
finally:
|
||||
os.unlink(tmp.name)
|
||||
else:
|
||||
config.load_incluster_config()
|
||||
|
||||
return client.CustomObjectsApi(), client.CoreV1Api(), client.AppsV1Api()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# CR / pod parsing helpers (pure — easy to unit test)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _network_from_namespace(namespace: str, chain: str) -> str:
|
||||
"""Derive network from a namespace like ``lux-mainnet`` → ``mainnet``."""
|
||||
prefix = f"{chain}-"
|
||||
if namespace.startswith(prefix):
|
||||
return namespace[len(prefix):] or "mainnet"
|
||||
# Fall back to a trailing known segment if present.
|
||||
for net in ("mainnet", "testnet", "devnet", "localnet"):
|
||||
if net in namespace:
|
||||
return net
|
||||
return namespace or "mainnet"
|
||||
|
||||
|
||||
def _cr_status(cr: dict) -> str:
|
||||
phase = (cr.get("status") or {}).get("phase", "Pending")
|
||||
return _PHASE_TO_STATUS.get(phase, "provisioning")
|
||||
|
||||
|
||||
def _cr_image(cr: dict, default: str) -> str:
|
||||
img = (cr.get("spec") or {}).get("image") or {}
|
||||
if isinstance(img, str):
|
||||
return img
|
||||
repo = img.get("repository")
|
||||
tag = img.get("tag")
|
||||
if repo:
|
||||
return f"{repo}:{tag}" if tag else repo
|
||||
return default
|
||||
|
||||
|
||||
def _pod_phase_to_status(phase: Optional[str], ready: bool) -> str:
|
||||
"""Translate a pod phase + readiness into a Node.status."""
|
||||
if phase == "Running":
|
||||
return "running" if ready else "syncing"
|
||||
if phase == "Pending":
|
||||
return "provisioning"
|
||||
if phase in ("Failed", "Unknown"):
|
||||
return "error"
|
||||
if phase == "Succeeded":
|
||||
return "stopped"
|
||||
return "provisioning"
|
||||
|
||||
|
||||
def _pod_ready(pod) -> bool:
|
||||
conds = getattr(pod.status, "conditions", None) or []
|
||||
return any(c.type == "Ready" and c.status == "True" for c in conds)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Detector
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class K8sNodeDetector:
|
||||
"""Discovers operator-managed validators on a cluster and adopts them.
|
||||
|
||||
Stateless apart from the DB session passed per call. Reusable for both the
|
||||
on-demand scan endpoints and the passive discovery path in ``list_nodes``.
|
||||
"""
|
||||
|
||||
async def scan_cluster(
|
||||
self,
|
||||
db: AsyncSession,
|
||||
cluster_id: str,
|
||||
kubeconfig: Optional[str],
|
||||
*,
|
||||
cluster_name: str = "",
|
||||
provider: str = "digitalocean",
|
||||
region: str = "sfo3",
|
||||
endpoint: Optional[str] = None,
|
||||
) -> ScanResult:
|
||||
"""Scan one cluster and upsert its CRs/pods as ``Node`` rows.
|
||||
|
||||
``cluster_id`` is the upstream id (DO cluster UUID, or ``"incluster"``)
|
||||
— used as ``Cluster.provider_id`` so re-scans adopt the same row.
|
||||
"""
|
||||
result = ScanResult(cluster_id=cluster_id)
|
||||
|
||||
try:
|
||||
custom_api, core_api, apps_api = _k8s_apis(kubeconfig)
|
||||
except Exception as exc: # noqa: BLE001 - surface, don't crash discovery
|
||||
logger.warning("k8s client init failed for cluster %s: %s", cluster_id, exc)
|
||||
result.errors.append(f"client: {exc}")
|
||||
return result
|
||||
|
||||
cluster = await self._upsert_cluster(
|
||||
db,
|
||||
provider_id=cluster_id,
|
||||
name=cluster_name or cluster_id,
|
||||
provider=provider,
|
||||
region=region,
|
||||
endpoint=endpoint,
|
||||
)
|
||||
result.cluster_db_id = str(cluster.id)
|
||||
|
||||
for spec in _CRDS:
|
||||
try:
|
||||
items = self._list_crs(custom_api, spec)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
logger.debug("No %s CRs on cluster %s: %s", spec.plural, cluster_id, exc)
|
||||
continue
|
||||
|
||||
for cr in items:
|
||||
result.crs_found += 1
|
||||
try:
|
||||
await self._adopt_cr(
|
||||
db, core_api, apps_api, cluster, spec, cr, region, result
|
||||
)
|
||||
except Exception as exc: # noqa: BLE001
|
||||
metadata = cr.get("metadata", {})
|
||||
ref = f"{metadata.get('namespace', '?')}/{metadata.get('name', '?')}"
|
||||
logger.warning("Adopt failed for %s %s: %s", spec.chain, ref, exc)
|
||||
result.errors.append(f"{spec.chain}/{ref}: {exc}")
|
||||
|
||||
await db.flush()
|
||||
return result
|
||||
|
||||
# ---- CR listing ----
|
||||
|
||||
@staticmethod
|
||||
def _list_crs(custom_api, spec: _CRDSpec) -> list[dict]:
|
||||
"""List a CRD across all namespaces (cluster-scoped call)."""
|
||||
resp = custom_api.list_cluster_custom_object(
|
||||
group=spec.group, version=spec.version, plural=spec.plural,
|
||||
)
|
||||
return resp.get("items", [])
|
||||
|
||||
# ---- pod enrichment ----
|
||||
|
||||
@staticmethod
|
||||
def _list_pods(core_api, namespace: str, cr_name: str) -> list:
|
||||
"""Best-effort pod list for a CR's workload in its namespace.
|
||||
|
||||
Operators label pods with the CR name; we accept the common label keys
|
||||
and otherwise fall back to all pods in the namespace whose name starts
|
||||
with the CR name (StatefulSet pods are ``<sts>-<ordinal>``).
|
||||
"""
|
||||
selectors = (
|
||||
f"app.kubernetes.io/instance={cr_name}",
|
||||
f"lux.network/network={cr_name}",
|
||||
f"zoo.network/chain={cr_name}",
|
||||
)
|
||||
for sel in selectors:
|
||||
try:
|
||||
pods = core_api.list_namespaced_pod(
|
||||
namespace=namespace, label_selector=sel
|
||||
).items
|
||||
if pods:
|
||||
return pods
|
||||
except Exception: # noqa: BLE001
|
||||
continue
|
||||
try:
|
||||
allp = core_api.list_namespaced_pod(namespace=namespace).items
|
||||
return [p for p in allp if (p.metadata.name or "").startswith(cr_name)]
|
||||
except Exception: # noqa: BLE001
|
||||
return []
|
||||
|
||||
@staticmethod
|
||||
def _cr_replicas(cr: dict) -> int:
|
||||
spec = cr.get("spec") or {}
|
||||
status = cr.get("status") or {}
|
||||
return (
|
||||
status.get("totalValidators")
|
||||
or status.get("totalNodes")
|
||||
or spec.get("validators")
|
||||
or spec.get("replicas")
|
||||
or 0
|
||||
)
|
||||
|
||||
async def _adopt_cr(
|
||||
self,
|
||||
db: AsyncSession,
|
||||
core_api,
|
||||
apps_api,
|
||||
cluster: Cluster,
|
||||
spec: _CRDSpec,
|
||||
cr: dict,
|
||||
region: str,
|
||||
result: ScanResult,
|
||||
) -> None:
|
||||
"""Upsert one CR's pods (or a synthetic node) as ``Node`` rows."""
|
||||
metadata = cr.get("metadata", {})
|
||||
cr_name = metadata.get("name", f"{spec.chain}d")
|
||||
namespace = metadata.get("namespace", f"{spec.chain}-mainnet")
|
||||
network = _network_from_namespace(namespace, spec.chain)
|
||||
cr_spec = cr.get("spec") or {}
|
||||
network_id = cr_spec.get("networkId")
|
||||
image = _cr_image(cr, spec.image_default)
|
||||
cr_status = _cr_status(cr)
|
||||
node_type = "validator" # operator CRs manage validators
|
||||
|
||||
pods = self._list_pods(core_api, namespace, cr_name)
|
||||
result.pods_seen += len(pods)
|
||||
|
||||
if pods:
|
||||
for pod in pods:
|
||||
pod_name = pod.metadata.name
|
||||
ready = _pod_ready(pod)
|
||||
status = _pod_phase_to_status(getattr(pod.status, "phase", None), ready)
|
||||
node_ip = getattr(pod.status, "pod_ip", None)
|
||||
await self._upsert_node(
|
||||
db,
|
||||
cluster=cluster,
|
||||
chain=spec.chain,
|
||||
network=network,
|
||||
pod_name=pod_name,
|
||||
name=pod_name,
|
||||
status=status,
|
||||
node_type=node_type,
|
||||
execution_client=spec.execution_client,
|
||||
image=image,
|
||||
region=region,
|
||||
network_id=network_id,
|
||||
pod_ip=node_ip,
|
||||
result=result,
|
||||
)
|
||||
else:
|
||||
# No pods visible (RBAC or headless) — still surface the CR as a
|
||||
# single logical node so the validator set is not invisible.
|
||||
await self._upsert_node(
|
||||
db,
|
||||
cluster=cluster,
|
||||
chain=spec.chain,
|
||||
network=network,
|
||||
pod_name=cr_name,
|
||||
name=cr_name,
|
||||
status=cr_status,
|
||||
node_type=node_type,
|
||||
execution_client=spec.execution_client,
|
||||
image=image,
|
||||
region=region,
|
||||
network_id=network_id,
|
||||
pod_ip=None,
|
||||
result=result,
|
||||
)
|
||||
|
||||
# ---- upserts ----
|
||||
|
||||
@staticmethod
|
||||
async def _upsert_cluster(
|
||||
db: AsyncSession,
|
||||
*,
|
||||
provider_id: str,
|
||||
name: str,
|
||||
provider: str,
|
||||
region: str,
|
||||
endpoint: Optional[str],
|
||||
) -> Cluster:
|
||||
"""Find-or-create the ``Cluster`` row keyed on ``provider_id``."""
|
||||
existing = (
|
||||
await db.execute(select(Cluster).where(Cluster.provider_id == provider_id))
|
||||
).scalar_one_or_none()
|
||||
if existing is not None:
|
||||
if endpoint and not existing.endpoint:
|
||||
existing.endpoint = endpoint
|
||||
return existing
|
||||
|
||||
cluster = Cluster(
|
||||
name=name,
|
||||
provider=provider,
|
||||
provider_id=provider_id,
|
||||
region=region,
|
||||
status="running",
|
||||
endpoint=endpoint,
|
||||
)
|
||||
db.add(cluster)
|
||||
await db.flush() # assigns cluster.id
|
||||
return cluster
|
||||
|
||||
@staticmethod
|
||||
async def _upsert_node(
|
||||
db: AsyncSession,
|
||||
*,
|
||||
cluster: Cluster,
|
||||
chain: str,
|
||||
network: str,
|
||||
pod_name: str,
|
||||
name: str,
|
||||
status: str,
|
||||
node_type: str,
|
||||
execution_client: str,
|
||||
image: str,
|
||||
region: str,
|
||||
network_id: Any,
|
||||
pod_ip: Optional[str],
|
||||
result: ScanResult,
|
||||
) -> None:
|
||||
"""Adopt-or-create a ``Node`` keyed on ``(cluster_id, pod_name)``.
|
||||
|
||||
Existing nodes are updated in place (status / metrics / image) — never
|
||||
recreated. New nodes are inserted.
|
||||
"""
|
||||
rpc_endpoint = None
|
||||
if pod_ip:
|
||||
port = 9650 if chain == "lux" else 8545
|
||||
rpc_endpoint = f"http://{pod_ip}:{port}/ext/bc/C/rpc"
|
||||
|
||||
existing = (
|
||||
await db.execute(
|
||||
select(Node).where(
|
||||
Node.cluster_id == cluster.id,
|
||||
Node.pod_name == pod_name,
|
||||
)
|
||||
)
|
||||
).scalar_one_or_none()
|
||||
|
||||
now = datetime.now(UTC)
|
||||
if existing is not None:
|
||||
# ADOPT: refresh mutable status, never duplicate.
|
||||
existing.status = status
|
||||
existing.chain = chain
|
||||
existing.network = network
|
||||
existing.node_type = node_type
|
||||
existing.execution_client = execution_client
|
||||
existing.image = image
|
||||
if rpc_endpoint:
|
||||
existing.rpc_endpoint = rpc_endpoint
|
||||
existing.last_health_check = now
|
||||
result.nodes_adopted += 1
|
||||
return
|
||||
|
||||
node = Node(
|
||||
name=name,
|
||||
chain=chain,
|
||||
network=network,
|
||||
provider="kubernetes",
|
||||
status=status,
|
||||
node_type=node_type,
|
||||
cluster_id=cluster.id,
|
||||
region=region,
|
||||
rpc_endpoint=rpc_endpoint,
|
||||
execution_client=execution_client,
|
||||
image=image,
|
||||
pod_name=pod_name,
|
||||
last_health_check=now,
|
||||
)
|
||||
db.add(node)
|
||||
result.nodes_created += 1
|
||||
|
||||
|
||||
# Module-level singleton — stateless, safe to share.
|
||||
detector = K8sNodeDetector()
|
||||
@@ -3,7 +3,9 @@
|
||||
from bootnode.db.models import (
|
||||
ApiKey,
|
||||
Base,
|
||||
Cluster,
|
||||
GasPolicy,
|
||||
Node,
|
||||
OrgCluster,
|
||||
Project,
|
||||
SmartWallet,
|
||||
@@ -23,7 +25,9 @@ __all__ = [
|
||||
"init_db",
|
||||
"Base",
|
||||
"ApiKey",
|
||||
"Cluster",
|
||||
"GasPolicy",
|
||||
"Node",
|
||||
"OrgCluster",
|
||||
"Project",
|
||||
"SmartWallet",
|
||||
|
||||
@@ -8,6 +8,7 @@ from sqlalchemy import (
|
||||
JSON,
|
||||
Boolean,
|
||||
DateTime,
|
||||
Float,
|
||||
ForeignKey,
|
||||
Integer,
|
||||
String,
|
||||
@@ -350,3 +351,111 @@ class OrgCluster(Base):
|
||||
|
||||
# Relationships
|
||||
project: Mapped["Project"] = relationship(back_populates="clusters")
|
||||
|
||||
|
||||
class Cluster(Base):
|
||||
"""Kubernetes cluster hosting blockchain nodes.
|
||||
|
||||
A physical/managed K8s cluster (DOKS, EKS, …) whose running blockchain
|
||||
workloads Bootnode discovers and adopts. Distinct from ``OrgCluster``
|
||||
(a project→cluster access-grant mapping): ``Cluster`` is the resource,
|
||||
``OrgCluster`` is the grant. ``provider_id`` is the upstream cluster id
|
||||
(e.g. the DigitalOcean cluster UUID) and is the natural key the k8s
|
||||
detector upserts against.
|
||||
"""
|
||||
|
||||
__tablename__ = "clusters"
|
||||
|
||||
id: Mapped[uuid.UUID] = mapped_column(
|
||||
Uuid(as_uuid=True), primary_key=True, default=uuid.uuid4
|
||||
)
|
||||
name: Mapped[str] = mapped_column(String(255), nullable=False)
|
||||
provider: Mapped[str] = mapped_column(String(50), nullable=False, default="digitalocean")
|
||||
provider_id: Mapped[str | None] = mapped_column(String(255), nullable=True, unique=True)
|
||||
region: Mapped[str] = mapped_column(String(50), nullable=False, default="sfo3")
|
||||
status: Mapped[str] = mapped_column(String(50), nullable=False, default="running")
|
||||
kubernetes_version: Mapped[str | None] = mapped_column(String(50), nullable=True)
|
||||
endpoint: Mapped[str | None] = mapped_column(String(500), nullable=True)
|
||||
node_count: Mapped[int] = mapped_column(Integer, default=3)
|
||||
node_size: Mapped[str] = mapped_column(String(100), default="s-4vcpu-8gb")
|
||||
created_at: Mapped[datetime] = mapped_column(
|
||||
DateTime(timezone=True), server_default=func.now()
|
||||
)
|
||||
updated_at: Mapped[datetime] = mapped_column(
|
||||
DateTime(timezone=True), server_default=func.now(), onupdate=func.now()
|
||||
)
|
||||
|
||||
# Relationships
|
||||
nodes: Mapped[list["Node"]] = relationship(
|
||||
back_populates="cluster", cascade="all, delete-orphan"
|
||||
)
|
||||
|
||||
|
||||
class Node(Base):
|
||||
"""A blockchain node — discovered (Docker/K8s) or provisioned by Bootnode.
|
||||
|
||||
Single persistent record behind ``GET /v1/nodes``. K8s-managed validators
|
||||
(luxd / zood StatefulSet pods adopted from LuxNetwork / ZooChain CRs) and
|
||||
Docker-managed nodes both land here; the ``provider`` field distinguishes
|
||||
them. ``(cluster_id, pod_name)`` and ``(provider, container_id)`` are the
|
||||
natural keys the detectors upsert against so re-scans ADOPT rather than
|
||||
duplicate.
|
||||
"""
|
||||
|
||||
__tablename__ = "nodes"
|
||||
|
||||
id: Mapped[uuid.UUID] = mapped_column(
|
||||
Uuid(as_uuid=True), primary_key=True, default=uuid.uuid4
|
||||
)
|
||||
name: Mapped[str] = mapped_column(String(255), nullable=False)
|
||||
chain: Mapped[str] = mapped_column(String(50), nullable=False)
|
||||
network: Mapped[str] = mapped_column(String(50), nullable=False, default="mainnet")
|
||||
provider: Mapped[str] = mapped_column(String(50), nullable=False, default="kubernetes")
|
||||
status: Mapped[str] = mapped_column(String(50), nullable=False, default="provisioning")
|
||||
node_type: Mapped[str] = mapped_column(String(50), nullable=False, default="rpc")
|
||||
|
||||
# Placement
|
||||
cluster_id: Mapped[uuid.UUID | None] = mapped_column(
|
||||
Uuid(as_uuid=True), ForeignKey("clusters.id", ondelete="SET NULL"), nullable=True
|
||||
)
|
||||
region: Mapped[str] = mapped_column(String(50), nullable=False, default="sfo3")
|
||||
latitude: Mapped[float | None] = mapped_column(Float, nullable=True)
|
||||
longitude: Mapped[float | None] = mapped_column(Float, nullable=True)
|
||||
|
||||
# Endpoints
|
||||
rpc_endpoint: Mapped[str | None] = mapped_column(String(500), nullable=True)
|
||||
ws_endpoint: Mapped[str | None] = mapped_column(String(500), nullable=True)
|
||||
beacon_endpoint: Mapped[str | None] = mapped_column(String(500), nullable=True)
|
||||
|
||||
# Clients / image
|
||||
execution_client: Mapped[str | None] = mapped_column(String(100), nullable=True)
|
||||
consensus_client: Mapped[str | None] = mapped_column(String(100), nullable=True)
|
||||
sync_mode: Mapped[str | None] = mapped_column(String(50), nullable=True)
|
||||
image: Mapped[str | None] = mapped_column(String(500), nullable=True)
|
||||
|
||||
# Runtime handles
|
||||
container_id: Mapped[str | None] = mapped_column(String(100), nullable=True)
|
||||
pod_name: Mapped[str | None] = mapped_column(String(255), nullable=True)
|
||||
|
||||
# Live metrics (snapshot; real-time lives in Redis/Prometheus)
|
||||
sync_progress: Mapped[float | None] = mapped_column(Float, nullable=True)
|
||||
peer_count: Mapped[int | None] = mapped_column(Integer, nullable=True)
|
||||
block_height: Mapped[int | None] = mapped_column(Integer, nullable=True)
|
||||
cpu_usage: Mapped[float | None] = mapped_column(Float, nullable=True)
|
||||
memory_usage_mb: Mapped[float | None] = mapped_column(Float, nullable=True)
|
||||
disk_usage_gb: Mapped[float | None] = mapped_column(Float, nullable=True)
|
||||
uptime_seconds: Mapped[int | None] = mapped_column(Integer, nullable=True)
|
||||
|
||||
error: Mapped[str | None] = mapped_column(Text, nullable=True)
|
||||
last_health_check: Mapped[datetime | None] = mapped_column(
|
||||
DateTime(timezone=True), nullable=True
|
||||
)
|
||||
created_at: Mapped[datetime] = mapped_column(
|
||||
DateTime(timezone=True), server_default=func.now()
|
||||
)
|
||||
updated_at: Mapped[datetime] = mapped_column(
|
||||
DateTime(timezone=True), server_default=func.now(), onupdate=func.now()
|
||||
)
|
||||
|
||||
# Relationships
|
||||
cluster: Mapped["Cluster | None"] = relationship(back_populates="nodes")
|
||||
|
||||
@@ -24,6 +24,7 @@ dependencies = [
|
||||
"pydantic-settings>=2.6.0",
|
||||
"sqlalchemy[asyncio]>=2.0.36",
|
||||
"asyncpg>=0.30.0",
|
||||
"aiosqlite>=0.20.0",
|
||||
"alembic>=1.14.0",
|
||||
"redis>=5.2.0",
|
||||
"httpx>=0.28.0",
|
||||
|
||||
Reference in New Issue
Block a user