feat: add IPAM discovery from Proxmox with IP enrichment, improve workload insights, and enhance DataTable interactivity
Add /ipam/discover endpoint to automatically import IP addresses from Proxmox clusters with error tracking and audit logging, implement ensure_discovered_network helper to create "discovered-ipam" network for auto-discovered IPs, add import_discovered_ips function to parse IP interfaces and create subnet/address records with assignment tracking, enhance ProxmoxProvider.enrich_work
This commit is contained in:
@@ -1,6 +1,7 @@
|
||||
from datetime import datetime
|
||||
import csv
|
||||
import io
|
||||
from ipaddress import ip_interface
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException
|
||||
from fastapi.responses import StreamingResponse
|
||||
@@ -93,6 +94,48 @@ def setup_setting(db: Session) -> SystemSetting:
|
||||
return setting
|
||||
|
||||
|
||||
def ensure_discovered_network(db: Session, cluster_id: str) -> Network:
|
||||
network = db.scalar(select(Network).where(Network.cluster_id == cluster_id, Network.name == "discovered-ipam"))
|
||||
if network:
|
||||
return network
|
||||
network = Network(
|
||||
cluster_id=cluster_id,
|
||||
name="discovered-ipam",
|
||||
kind="discovered",
|
||||
description="Automatically created for IP addresses discovered during Proxmox sync.",
|
||||
)
|
||||
db.add(network)
|
||||
db.flush()
|
||||
return network
|
||||
|
||||
|
||||
def import_discovered_ips(db: Session, cluster_id: str, workload: Workload, addresses: list[str]) -> int:
|
||||
imported = 0
|
||||
for value in addresses:
|
||||
try:
|
||||
interface = ip_interface(value)
|
||||
except ValueError:
|
||||
continue
|
||||
if interface.ip.is_loopback or interface.ip.is_link_local:
|
||||
continue
|
||||
network = ensure_discovered_network(db, cluster_id)
|
||||
cidr = str(interface.network)
|
||||
subnet = db.scalar(select(Subnet).where(Subnet.network_id == network.id, Subnet.cidr == cidr))
|
||||
if not subnet:
|
||||
subnet = Subnet(network_id=network.id, cidr=cidr)
|
||||
db.add(subnet)
|
||||
db.flush()
|
||||
address_value = str(interface.ip)
|
||||
existing = db.scalar(select(IpAddress).where(IpAddress.subnet_id == subnet.id, IpAddress.address == address_value))
|
||||
if existing:
|
||||
existing.workload_id = workload.id
|
||||
existing.status = "assigned"
|
||||
else:
|
||||
db.add(IpAddress(subnet_id=subnet.id, address=address_value, status="assigned", workload_id=workload.id))
|
||||
imported += 1
|
||||
return imported
|
||||
|
||||
|
||||
@api_router.get("/setup/status", response_model=SetupStatus)
|
||||
def setup_status(db: Session = Depends(get_db)) -> SetupStatus:
|
||||
setting = setup_setting(db)
|
||||
@@ -171,10 +214,7 @@ def dashboard(_: CurrentUser, db: Session = Depends(get_db)) -> dict:
|
||||
{"id": node.id, "name": node.name, "status": node.status, "cluster_id": node.cluster_id}
|
||||
for node in faulty_nodes
|
||||
],
|
||||
"top_talkers": [
|
||||
{"name": "finance-app-2", "bytes": 942000000},
|
||||
{"name": "core-services-1", "bytes": 512000000},
|
||||
],
|
||||
"top_talkers": [],
|
||||
}
|
||||
|
||||
|
||||
@@ -326,6 +366,7 @@ async def sync_cluster(cluster_id: str, user: CurrentUser, db: Session = Depends
|
||||
workload.name = raw_workload.get("name") or workload.name
|
||||
workload.kind = raw_workload.get("type") or raw_workload.get("kind") or workload.kind
|
||||
workload.status = raw_workload.get("status") or workload.status
|
||||
import_discovered_ips(db, cluster.id, workload, raw_workload.get("ip_addresses", []))
|
||||
|
||||
network_by_name = {
|
||||
network.name: network
|
||||
@@ -367,35 +408,29 @@ def workload_insights(workload_id: str, _: CurrentUser, db: Session = Depends(ge
|
||||
policies = db.scalars(
|
||||
select(Policy).where((Policy.project_id == workload.project_id) | (Policy.project_id.is_(None))).order_by(Policy.name)
|
||||
).all()
|
||||
traffic = [
|
||||
{
|
||||
"timestamp": datetime.utcnow().isoformat(),
|
||||
"source": workload.name,
|
||||
"destination": "finance-db-1" if "web" in workload.tags else "core-services-1",
|
||||
"protocol": "tcp",
|
||||
"port": 5432 if "web" in workload.tags else 22,
|
||||
"bytes": 1489200,
|
||||
"decision": "allowed",
|
||||
},
|
||||
{
|
||||
"timestamp": datetime.utcnow().isoformat(),
|
||||
"source": "unknown-external",
|
||||
"destination": workload.name,
|
||||
"protocol": "tcp",
|
||||
"port": 3389,
|
||||
"bytes": 22140,
|
||||
"decision": "would_block",
|
||||
},
|
||||
]
|
||||
assigned_ips = db.scalars(select(IpAddress).where(IpAddress.workload_id == workload.id)).all()
|
||||
traffic = []
|
||||
audit_mode_notes = [
|
||||
f"{policy.name} is in audit mode; matching traffic is logged without enforcement."
|
||||
for policy in policies
|
||||
if policy.enforcement_mode == "audit"
|
||||
]
|
||||
decision = "audit" if audit_mode_notes else "allowed"
|
||||
decision = "audit" if audit_mode_notes else "unknown"
|
||||
return WorkloadInsight(
|
||||
workload=workload,
|
||||
traffic=traffic,
|
||||
traffic=[
|
||||
{
|
||||
"source": workload.name,
|
||||
"destination": "unknown",
|
||||
"protocol": "unknown",
|
||||
"port": "unknown",
|
||||
"bytes": 0,
|
||||
"decision": "no_flow_telemetry",
|
||||
"ip_addresses": [address.address for address in assigned_ips],
|
||||
}
|
||||
]
|
||||
if assigned_ips
|
||||
else traffic,
|
||||
matching_policies=policies,
|
||||
effective_decision=decision,
|
||||
audit_mode_notes=audit_mode_notes,
|
||||
@@ -441,6 +476,38 @@ def ipam_addresses(_: CurrentUser, db: Session = Depends(get_db)) -> list[IpAddr
|
||||
return db.scalars(select(IpAddress).order_by(IpAddress.address)).all()
|
||||
|
||||
|
||||
@api_router.post("/ipam/discover")
|
||||
async def discover_ipam(user: CurrentUser, db: Session = Depends(get_db)) -> dict:
|
||||
imported = 0
|
||||
errors = []
|
||||
clusters = db.scalars(select(Cluster).order_by(Cluster.name)).all()
|
||||
for cluster in clusters:
|
||||
try:
|
||||
inventory = await get_provider(cluster.provider).sync_inventory(
|
||||
ProviderConnection(
|
||||
api_url=cluster.api_url,
|
||||
token=cluster.token_ref or "",
|
||||
verify_tls=cluster.verify_tls,
|
||||
read_only=True,
|
||||
)
|
||||
)
|
||||
workload_by_external_id = {
|
||||
workload.external_id: workload
|
||||
for workload in db.scalars(select(Workload).where(Workload.cluster_id == cluster.id)).all()
|
||||
}
|
||||
for raw_workload in inventory.get("workloads", []):
|
||||
external_id = str(raw_workload.get("vmid") or raw_workload.get("id") or "")
|
||||
workload = workload_by_external_id.get(external_id)
|
||||
if workload:
|
||||
imported += import_discovered_ips(db, cluster.id, workload, raw_workload.get("ip_addresses", []))
|
||||
except Exception as exc:
|
||||
errors.append({"cluster": cluster.name, "error": str(exc)})
|
||||
db.add(Job(kind="ipam.discover", status="success" if not errors else "failed", progress=100, logs=[f"Imported {imported} IP addresses"], error=str(errors) if errors else None))
|
||||
commit_or_400(db)
|
||||
write_audit(db, action="ipam.discover", object_type="ipam", user_id=user.id, new_values={"imported": imported, "errors": errors}, result="success" if not errors else "failed")
|
||||
return {"imported": imported, "errors": errors}
|
||||
|
||||
|
||||
@api_router.post("/ipam/addresses", response_model=IpAddressRead)
|
||||
def reserve_ip(payload: IpReservationCreate, user: CurrentUser, db: Session = Depends(get_db)) -> IpAddress:
|
||||
if not db.get(Subnet, payload.subnet_id):
|
||||
|
||||
Reference in New Issue
Block a user