feat: enhance Lux fleet management with in-cluster + StatefulSet discovery

- Add StatefulSet discovery fallback for kubectl-deployed fleets (non-Helm)
- Add in-cluster mode: auto-detect pod ServiceAccount, skip kubeconfig fetch
- Add list_fleets_incluster() for direct K8s API fleet discovery
- Add get_statefulsets() to HelmDeployer for kubectl-based fleet listing
- Make helm_revision optional (null for non-Helm fleets)
- Add Lux devnet network (chain_id 96370), fix testnet chain_id (96368)
- Add in-cluster self-detection for /v1/infra/clusters endpoint
- Fix kubeconfig fetch: use raw HTTP (pydo can't deserialize YAML), doctl fallback
- Install kubectl in API Dockerfile for fleet management operations
This commit is contained in:
2026-02-12 10:21:01 -08:00
parent 7acebc977d
commit 5d63821793
7 changed files with 433 additions and 58 deletions
+4 -1
View File
@@ -3,10 +3,13 @@ FROM python:3.12-slim
WORKDIR /app
# Install system dependencies
# Install system dependencies + kubectl for fleet management
RUN apt-get update && apt-get install -y --no-install-recommends \
build-essential \
curl \
&& curl -LO "https://dl.k8s.io/release/$(curl -L -s https://dl.k8s.io/release/stable.txt)/bin/linux/amd64/kubectl" \
&& install -o root -g root -m 0755 kubectl /usr/local/bin/kubectl \
&& rm kubectl \
&& rm -rf /var/lib/apt/lists/*
# Install uv for faster dependency installation
+60 -6
View File
@@ -248,12 +248,43 @@ async def do_create_cluster(client, cluster: ClusterCreate) -> Dict:
async def do_get_kubeconfig(client, cluster_id: str) -> str:
"""Get DOKS cluster kubeconfig"""
"""Get DOKS cluster kubeconfig via DO API.
Uses raw HTTP because pydo can't deserialize YAML responses.
Falls back to doctl CLI if available.
"""
import asyncio
import httpx
token = InfraConfig.DO_TOKEN
if not token:
raise HTTPException(status_code=500, detail="DO_TOKEN not configured")
# Try raw HTTP first (pydo can't handle YAML content-type)
try:
resp = client.kubernetes.get_kubeconfig(cluster_id=cluster_id)
return resp
except Exception as e:
raise HTTPException(status_code=500, detail=f"DO API error: {e}")
async with httpx.AsyncClient(timeout=30) as http:
resp = await http.get(
f"https://api.digitalocean.com/v2/kubernetes/clusters/{cluster_id}/kubeconfig",
headers={"Authorization": f"Bearer {token}"},
)
if resp.status_code == 200:
return resp.text
raise HTTPException(status_code=resp.status_code, detail=f"DO API: {resp.text[:200]}")
except httpx.HTTPError as e:
# Fallback to doctl CLI
try:
proc = await asyncio.create_subprocess_exec(
"doctl", "kubernetes", "cluster", "kubeconfig", "show", cluster_id,
"--access-token", token,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=30)
if proc.returncode == 0:
return stdout.decode()
raise HTTPException(status_code=502, detail=f"doctl error: {stderr.decode()[:200]}")
except FileNotFoundError:
raise HTTPException(status_code=500, detail=f"DO API error: {e}")
async def do_list_volumes(client) -> List[Dict]:
@@ -465,10 +496,33 @@ MOCK_VOLUMES = [
@router.get("/clusters", response_model=List[ClusterResponse])
async def list_clusters():
"""List all Kubernetes clusters"""
"""List all Kubernetes clusters.
Returns DO clusters if DO_TOKEN is set, otherwise returns the local
in-cluster entry when running inside K8s.
"""
client = get_do_client()
if client:
return await do_list_clusters(client)
# Detect in-cluster and return self
if os.path.exists("/var/run/secrets/kubernetes.io/serviceaccount/token"):
return [{
"id": "incluster",
"name": "Local Cluster",
"provider": "kubernetes",
"region": os.getenv("DO_REGION", "sfo3"),
"status": "running",
"kubernetes_version": "",
"node_count": 0,
"node_size": "",
"endpoint": "https://kubernetes.default.svc",
"created_at": "",
"updated_at": None,
"node_pools": [],
"tags": ["incluster"],
}]
return MOCK_CLUSTERS
+55 -7
View File
@@ -2,9 +2,14 @@
Provides CRUD operations for Lux blockchain validator fleets deployed
via Helm charts to DOKS clusters.
Supports two modes:
1. Remote: cluster_id provided → fetch kubeconfig from DigitalOcean API
2. In-cluster: no cluster_id or no DO token → use pod ServiceAccount
"""
import logging
import os
import re
from typing import Optional
@@ -36,6 +41,9 @@ _manager = LuxFleetManager()
# K8s name validation: RFC 1123 DNS label
_K8S_NAME_RE = re.compile(r"^[a-z0-9]([a-z0-9\-]{0,61}[a-z0-9])?$")
# In-cluster sentinel -- used when no cluster_id is provided
_INCLUSTER_ID = "incluster"
def _validate_k8s_name(name: str, field: str = "name") -> None:
"""Validate a string is a valid K8s name (RFC 1123 DNS label)."""
@@ -43,20 +51,46 @@ def _validate_k8s_name(name: str, field: str = "name") -> None:
raise HTTPException(400, f"Invalid {field}: must match RFC 1123 DNS label")
async def _get_kubeconfig(cluster_id: str) -> str:
"""Fetch kubeconfig for a DOKS cluster via the infra API."""
def _has_do_client() -> bool:
"""Check if DigitalOcean client is available."""
try:
from bootnode.api.infra import get_do_client
return get_do_client() is not None
except Exception:
return False
def _is_incluster() -> bool:
"""Detect if running inside a Kubernetes pod."""
return os.path.exists("/var/run/secrets/kubernetes.io/serviceaccount/token")
async def _get_kubeconfig(cluster_id: str) -> Optional[str]:
"""Fetch kubeconfig for a DOKS cluster via the infra API.
Returns None if cluster_id is the in-cluster sentinel.
"""
if cluster_id == _INCLUSTER_ID:
return None
try:
from bootnode.api.infra import do_get_kubeconfig, get_do_client
client = get_do_client()
if not client:
if _is_incluster():
return None # fall back to in-cluster
raise HTTPException(500, "DigitalOcean client not configured")
return await do_get_kubeconfig(client, cluster_id)
except ImportError:
if _is_incluster():
return None
raise HTTPException(500, "Infrastructure module not available")
except HTTPException:
raise
except Exception:
logger.error("Failed to fetch kubeconfig for cluster %s", cluster_id)
if _is_incluster():
return None
raise HTTPException(502, "Failed to fetch kubeconfig for cluster")
@@ -87,10 +121,19 @@ async def create_fleet(
@router.get("/fleets", response_model=list[LuxFleetSummary])
async def list_fleets(_api_key: ApiKeyDep, cluster_id: Optional[str] = None) -> list[LuxFleetSummary]:
"""List all Lux fleets, optionally filtered by cluster."""
if cluster_id:
kubeconfig = await _get_kubeconfig(cluster_id)
return await _manager.list_fleets(kubeconfig, cluster_id)
"""List all Lux fleets, optionally filtered by cluster.
When running in-cluster and no cluster_id is provided, discovers
fleets on the local cluster using the pod's ServiceAccount.
"""
effective_cluster_id = cluster_id or (_INCLUSTER_ID if _is_incluster() else None)
if effective_cluster_id:
kubeconfig = await _get_kubeconfig(effective_cluster_id)
if kubeconfig is None:
# In-cluster mode
return await _manager.list_fleets_incluster(effective_cluster_id)
return await _manager.list_fleets(kubeconfig, effective_cluster_id)
# List across all known clusters
all_fleets = []
@@ -98,7 +141,10 @@ async def list_fleets(_api_key: ApiKeyDep, cluster_id: Optional[str] = None) ->
for cid in cluster_ids:
try:
kubeconfig = await _get_kubeconfig(cid)
fleets = await _manager.list_fleets(kubeconfig, cid)
if kubeconfig is None:
fleets = await _manager.list_fleets_incluster(cid)
else:
fleets = await _manager.list_fleets(kubeconfig, cid)
all_fleets.extend(fleets)
except Exception as e:
logger.warning("Failed to list fleets on cluster %s: %s", cid, e)
@@ -117,6 +163,8 @@ async def get_fleet(cluster_id: str, network: LuxNetwork, _api_key: ApiKeyDep) -
name = fdata["name"]
break
if kubeconfig is None:
return await _manager.get_status_incluster(name, network, cluster_id)
return await _manager.get_status(kubeconfig, name, network, cluster_id)
+276 -42
View File
@@ -81,6 +81,16 @@ class LuxFleetManager:
deployer._temp_files.append(kc_path)
return deployer
def _get_k8s_client(self):
"""Get a kubernetes client using in-cluster config."""
try:
from kubernetes import client, config
config.load_incluster_config()
return client.CoreV1Api(), client.AppsV1Api()
except Exception:
logger.warning("Failed to load in-cluster k8s config")
raise
def _get_values_files(self, network: LuxNetwork) -> list[Path]:
"""Get the network-specific values file if one exists."""
filename = NETWORK_VALUES_FILES.get(network.value)
@@ -248,10 +258,20 @@ class LuxFleetManager:
namespace = self._namespace(network)
net_cfg = NETWORK_CONFIGS[network]
# Get Helm release status
# Get Helm release status (optional -- fleet may be kubectl-deployed)
helm_revision = None
try:
release = await deployer.status(release_name, namespace)
helm_revision = release.revision
except HelmError:
pass # Not a Helm-managed fleet, continue with kubectl
# Get pods
pods = await deployer.get_pods(namespace, "app=luxd")
ready_count = sum(1 for p in pods if p.ready)
total = len(pods)
if total == 0 and helm_revision is None:
return LuxFleetResponse(
id=self._fleet_id(name, network),
name=name,
@@ -260,14 +280,9 @@ class LuxFleetManager:
status=LuxFleetStatus.ERROR,
replicas=0,
namespace=namespace,
error="Helm release not found",
error="No fleet found (no Helm release or pods)",
)
# Get pods
pods = await deployer.get_pods(namespace, "app=luxd")
ready_count = sum(1 for p in pods if p.ready)
total = len(pods)
# Get services for endpoint discovery
services = await deployer.get_services(namespace)
rpc_endpoint = None
@@ -317,7 +332,7 @@ class LuxFleetManager:
replicas=total,
ready_replicas=ready_count,
namespace=namespace,
helm_revision=release.revision,
helm_revision=helm_revision,
rpc_endpoint=rpc_endpoint,
staking_endpoint=staking_endpoint,
nodes=nodes,
@@ -330,51 +345,270 @@ class LuxFleetManager:
kubeconfig: str,
cluster_id: str,
) -> list[LuxFleetSummary]:
"""List all Lux fleets on a cluster."""
"""List all Lux fleets on a cluster.
Discovers fleets via Helm releases first, then falls back to
direct StatefulSet discovery for fleets deployed via kubectl.
"""
deployer = await self._get_deployer(kubeconfig)
try:
releases = await deployer.list_releases(
all_namespaces=True,
filter_pattern="^luxd-",
)
fleets = []
for r in releases:
# Infer network from release name (luxd-mainnet -> mainnet)
network_str = r.name.removeprefix("luxd-")
try:
network = LuxNetwork(network_str)
except ValueError:
seen_namespaces: set[str] = set()
# 1. Try Helm releases
try:
releases = await deployer.list_releases(
all_namespaces=True,
filter_pattern="^luxd-",
)
for r in releases:
network_str = r.name.removeprefix("luxd-")
try:
network = LuxNetwork(network_str)
except ValueError:
continue
pods = await deployer.get_pods(r.namespace, "app=luxd")
ready = sum(1 for p in pods if p.ready)
total = len(pods)
seen_namespaces.add(r.namespace)
if total == 0:
status = LuxFleetStatus.DEPLOYING
elif ready == total:
status = LuxFleetStatus.RUNNING
elif ready > 0:
status = LuxFleetStatus.DEGRADED
else:
status = LuxFleetStatus.ERROR
fleets.append(LuxFleetSummary(
id=f"{cluster_id}:{r.name}",
name=r.name,
network=network,
status=status,
replicas=total,
ready_replicas=ready,
cluster_id=cluster_id,
created_at=r.updated or "",
))
except Exception:
logger.debug("No Helm releases found, trying StatefulSet discovery")
# 2. Fall back to StatefulSet discovery for kubectl-deployed fleets
for network, cfg in NETWORK_CONFIGS.items():
if cfg.namespace in seen_namespaces:
continue
try:
sts_list = await deployer.get_statefulsets(namespace=cfg.namespace)
for sts in sts_list:
if sts["name"] != "luxd":
continue
total = sts["replicas"]
ready = sts["ready_replicas"]
pods = await deployer.get_pods(r.namespace, "app=luxd")
ready = sum(1 for p in pods if p.ready)
total = len(pods)
if total == 0:
status = LuxFleetStatus.DEPLOYING
elif ready == total:
status = LuxFleetStatus.RUNNING
elif ready > 0:
status = LuxFleetStatus.DEGRADED
else:
status = LuxFleetStatus.ERROR
if total == 0:
status = LuxFleetStatus.DEPLOYING
elif ready == total:
status = LuxFleetStatus.RUNNING
elif ready > 0:
status = LuxFleetStatus.DEGRADED
else:
status = LuxFleetStatus.ERROR
fleets.append(LuxFleetSummary(
id=f"{cluster_id}:{r.name}",
name=r.name,
network=network,
status=status,
replicas=total,
ready_replicas=ready,
cluster_id=cluster_id,
created_at=r.updated or "",
))
fleets.append(LuxFleetSummary(
id=f"{cluster_id}:luxd-{network.value}",
name=f"luxd-{network.value}",
network=network,
status=status,
replicas=total,
ready_replicas=ready,
cluster_id=cluster_id,
created_at=sts.get("created_at", ""),
))
except Exception:
continue
return fleets
finally:
deployer.cleanup_temp_files()
async def list_fleets_incluster(
self,
cluster_id: str,
) -> list[LuxFleetSummary]:
"""List all Lux fleets using in-cluster k8s access.
Discovers fleets via StatefulSet discovery in known namespaces.
"""
core_v1, apps_v1 = self._get_k8s_client()
fleets = []
for network, cfg in NETWORK_CONFIGS.items():
try:
sts_list = apps_v1.list_namespaced_stateful_set(
namespace=cfg.namespace,
label_selector="app=luxd",
)
for sts in sts_list.items:
total = sts.spec.replicas or 0
ready = sts.status.ready_replicas or 0
if total == 0:
fleet_status = LuxFleetStatus.DEPLOYING
elif ready == total:
fleet_status = LuxFleetStatus.RUNNING
elif ready > 0:
fleet_status = LuxFleetStatus.DEGRADED
else:
fleet_status = LuxFleetStatus.ERROR
created = ""
if sts.metadata.creation_timestamp:
created = sts.metadata.creation_timestamp.isoformat()
fleets.append(LuxFleetSummary(
id=f"{cluster_id}:luxd-{network.value}",
name=f"luxd-{network.value}",
network=network,
status=fleet_status,
replicas=total,
ready_replicas=ready,
cluster_id=cluster_id,
created_at=created,
))
# If no statefulset found, check for pods directly
if not sts_list.items:
pods = core_v1.list_namespaced_pod(
namespace=cfg.namespace,
label_selector="app=luxd",
)
if pods.items:
total = len(pods.items)
ready = sum(
1 for p in pods.items
if p.status.phase == "Running"
and p.status.container_statuses
and all(c.ready for c in p.status.container_statuses)
)
fleet_status = LuxFleetStatus.RUNNING if ready == total else (
LuxFleetStatus.DEGRADED if ready > 0 else LuxFleetStatus.ERROR
)
fleets.append(LuxFleetSummary(
id=f"{cluster_id}:luxd-{network.value}",
name=f"luxd-{network.value}",
network=network,
status=fleet_status,
replicas=total,
ready_replicas=ready,
cluster_id=cluster_id,
created_at="",
))
except Exception as e:
logger.warning("Failed to list fleet in %s: %s", cfg.namespace, e)
continue
return fleets
async def get_status_incluster(
self,
name: str,
network: LuxNetwork,
cluster_id: str,
) -> LuxFleetResponse:
"""Get full fleet status using in-cluster k8s access."""
core_v1, apps_v1 = self._get_k8s_client()
namespace = self._namespace(network)
net_cfg = NETWORK_CONFIGS[network]
# Get pods
pod_list = core_v1.list_namespaced_pod(
namespace=namespace,
label_selector="app=luxd",
)
total = len(pod_list.items)
if total == 0:
return LuxFleetResponse(
id=self._fleet_id(name, network),
name=name,
cluster_id=cluster_id,
network=network,
status=LuxFleetStatus.ERROR,
replicas=0,
namespace=namespace,
error="No pods found",
)
ready_count = 0
for p in pod_list.items:
if (p.status.phase == "Running"
and p.status.container_statuses
and all(c.ready for c in p.status.container_statuses)):
ready_count += 1
# Get services for endpoint discovery
svc_list = core_v1.list_namespaced_service(namespace=namespace)
rpc_endpoint = None
staking_endpoint = None
for svc in svc_list.items:
if svc.status.load_balancer and svc.status.load_balancer.ingress:
for ing in svc.status.load_balancer.ingress:
ext_ip = ing.ip
if ext_ip and svc.spec.ports:
for port in svc.spec.ports:
if port.port == net_cfg.http_port:
rpc_endpoint = f"http://{ext_ip}:{net_cfg.http_port}"
elif port.port == net_cfg.staking_port:
staking_endpoint = f"{ext_ip}:{net_cfg.staking_port}"
# Build node info
nodes = []
sorted_pods = sorted(pod_list.items, key=lambda p: p.metadata.name)
for i, pod in enumerate(sorted_pods):
pod_ready = (
pod.status.phase == "Running"
and pod.status.container_statuses
and all(c.ready for c in pod.status.container_statuses)
)
if pod_ready:
node_status = LuxNodeStatus.HEALTHY
elif pod.status.phase == "Pending":
node_status = LuxNodeStatus.PENDING
else:
node_status = LuxNodeStatus.UNHEALTHY
nodes.append(LuxNodeInfo(
pod_name=pod.metadata.name,
pod_index=i,
status=node_status,
http_port=net_cfg.http_port,
staking_port=net_cfg.staking_port,
))
# Determine fleet status
if ready_count == total:
fleet_status = LuxFleetStatus.RUNNING
elif ready_count > 0:
fleet_status = LuxFleetStatus.DEGRADED
else:
fleet_status = LuxFleetStatus.ERROR
return LuxFleetResponse(
id=self._fleet_id(name, network),
name=name,
cluster_id=cluster_id,
network=network,
status=fleet_status,
replicas=total,
ready_replicas=ready_count,
namespace=namespace,
rpc_endpoint=rpc_endpoint,
staking_endpoint=staking_endpoint,
nodes=nodes,
)
async def get_pod_logs(
self,
kubeconfig: str,
+1 -1
View File
@@ -292,7 +292,7 @@ class LuxFleetResponse(BaseModel):
ready_replicas: int = 0
image: ImageConfig = Field(default_factory=ImageConfig)
namespace: str = ""
helm_revision: int = 0
helm_revision: Optional[int] = None
rpc_endpoint: Optional[str] = None
staking_endpoint: Optional[str] = None
nodes: list[LuxNodeInfo] = Field(default_factory=list)
+9 -1
View File
@@ -249,12 +249,20 @@ class ChainRegistry:
),
"testnet": Network(
name="Testnet",
chain_id=96370,
chain_id=96368,
is_testnet=True,
rpc_env_var="lux_testnet_rpc",
explorer_url="https://explore.testnet.lux.network",
native_currency="LUX",
),
"devnet": Network(
name="Devnet",
chain_id=96370,
is_testnet=True,
rpc_env_var="lux_devnet_rpc",
explorer_url="https://explore.devnet.lux.network",
native_currency="LUX",
),
},
)
+28
View File
@@ -363,6 +363,34 @@ class HelmDeployer:
))
return pods
async def get_statefulsets(
self, namespace: str = "", all_namespaces: bool = False, label_selector: str = ""
) -> list[dict]:
"""kubectl get statefulsets."""
args = self._base_kubectl_args() + ["get", "statefulsets", "--output", "json"]
if all_namespaces:
args.append("--all-namespaces")
elif namespace:
args += ["--namespace", namespace]
if label_selector:
args += ["-l", label_selector]
out = await self._run(args)
data = json.loads(out)
results = []
for item in data.get("items", []):
meta = item.get("metadata", {})
spec = item.get("spec", {})
status = item.get("status", {})
results.append({
"name": meta.get("name", ""),
"namespace": meta.get("namespace", ""),
"replicas": spec.get("replicas", 0),
"ready_replicas": status.get("readyReplicas", 0),
"created_at": meta.get("creationTimestamp", ""),
})
return results
async def get_services(self, namespace: str, label_selector: str = "") -> list[dict]:
"""kubectl get services, returns raw dicts."""
args = self._base_kubectl_args() + [