nessi
67eee0662a
Add RuntimeSettingsRead/RuntimeSettingsUpdate schemas with auto_node_sync_enabled/auto_node_sync_interval_minutes/auto_ipam_sync_enabled/auto_ipam_sync_interval_minutes/last_node_auto_sync_at/last_ipam_auto_sync_at fields, implement firewall_log_lines_from_files to parse /var/log/pve-firewall.log with max_lines limit and error collection, extend collect_firewall_log
236 lines
10 KiB
Python
236 lines
10 KiB
Python
from datetime import datetime, timedelta
|
|
from ipaddress import ip_interface
|
|
from typing import Any
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.domain import Cluster, IpAddress, Job, Network, Node, Subnet, SystemSetting, Workload
|
|
from app.services.providers.base import ProviderConnection
|
|
from app.services.providers.registry import get_provider
|
|
|
|
|
|
def runtime_setting(db: Session) -> SystemSetting:
|
|
setting = db.get(SystemSetting, "runtime")
|
|
if not setting:
|
|
setting = SystemSetting(key="runtime", value={})
|
|
db.add(setting)
|
|
db.commit()
|
|
db.refresh(setting)
|
|
return setting
|
|
|
|
|
|
def parse_last_run(value: dict[str, Any], key: str) -> datetime | None:
|
|
raw = value.get(key)
|
|
if not raw:
|
|
return None
|
|
try:
|
|
return datetime.fromisoformat(str(raw))
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def due(value: dict[str, Any], enabled_key: str, interval_key: str, last_key: str) -> bool:
|
|
if not bool(value.get(enabled_key, False)):
|
|
return False
|
|
interval = int(value.get(interval_key) or 60)
|
|
last_run = parse_last_run(value, last_key)
|
|
return last_run is None or datetime.utcnow() - last_run >= timedelta(minutes=interval)
|
|
|
|
|
|
def is_container_network(value: str) -> bool:
|
|
try:
|
|
interface = ip_interface(value)
|
|
except ValueError:
|
|
return False
|
|
ip = interface.ip
|
|
network = str(interface.network)
|
|
if ip.is_loopback or ip.is_link_local:
|
|
return True
|
|
if ip.version == 4 and ip.packed[0] == 172 and 17 <= ip.packed[1] <= 31:
|
|
return True
|
|
return network.startswith(("10.42.", "10.43.", "10.244.", "10.245."))
|
|
|
|
|
|
def cleanup_discovered_container_networks(db: Session) -> int:
|
|
removed = 0
|
|
discovered_networks = db.scalars(select(Network).where(Network.name == "discovered-ipam")).all()
|
|
for network in discovered_networks:
|
|
subnets = db.scalars(select(Subnet).where(Subnet.network_id == network.id)).all()
|
|
for subnet in subnets:
|
|
if is_container_network(subnet.cidr):
|
|
addresses = db.scalars(select(IpAddress).where(IpAddress.subnet_id == subnet.id)).all()
|
|
for address in addresses:
|
|
db.delete(address)
|
|
removed += 1
|
|
db.delete(subnet)
|
|
return removed
|
|
|
|
|
|
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 is_container_network(value):
|
|
continue
|
|
network = ensure_discovered_network(db, cluster_id)
|
|
subnet = db.scalar(select(Subnet).where(Subnet.network_id == network.id, Subnet.cidr == str(interface.network)))
|
|
if not subnet:
|
|
subnet = Subnet(network_id=network.id, cidr=str(interface.network))
|
|
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
|
|
|
|
|
|
async def sync_cluster_inventory(db: Session, cluster: Cluster, job_kind: str = "proxmox.auto_sync") -> dict[str, Any]:
|
|
provider = get_provider(cluster.provider)
|
|
try:
|
|
inventory = await provider.sync_inventory(
|
|
ProviderConnection(
|
|
api_url=cluster.api_url,
|
|
token=cluster.token_ref or "",
|
|
verify_tls=cluster.verify_tls,
|
|
read_only=cluster.mode == "read_only",
|
|
)
|
|
)
|
|
except Exception as exc:
|
|
cluster.last_sync_at = datetime.utcnow()
|
|
cluster.last_sync_status = "failed"
|
|
cluster.last_sync_error = str(exc)
|
|
db.add(Job(kind=job_kind, status="failed", progress=100, logs=[f"Auto sync failed for {cluster.name}"], error=str(exc)))
|
|
db.commit()
|
|
return {"cluster": cluster.name, "status": "failed", "error": str(exc)}
|
|
|
|
cluster.last_sync_at = datetime.utcnow()
|
|
cluster.last_sync_status = "success"
|
|
cluster.last_sync_error = None
|
|
node_by_name = {node.name: node for node in db.scalars(select(Node).where(Node.cluster_id == cluster.id)).all()}
|
|
for raw_node in inventory.get("nodes", []):
|
|
name = raw_node.get("node") or raw_node.get("name")
|
|
if not name:
|
|
continue
|
|
node = node_by_name.get(name)
|
|
if not node:
|
|
node = Node(cluster_id=cluster.id, name=name)
|
|
db.add(node)
|
|
node_by_name[name] = node
|
|
node.status = raw_node.get("status", node.status)
|
|
node.cpu_count = int(raw_node.get("maxcpu") or raw_node.get("cpu_count") or node.cpu_count or 0)
|
|
maxmem = raw_node.get("maxmem")
|
|
node.memory_mb = int(maxmem / 1024 / 1024) if isinstance(maxmem, int | float) else int(raw_node.get("memory_mb") or node.memory_mb or 0)
|
|
|
|
db.flush()
|
|
workloads = {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 "")
|
|
if not external_id:
|
|
continue
|
|
node = node_by_name.get(raw_workload.get("node")) or next(iter(node_by_name.values()), None)
|
|
if not node:
|
|
continue
|
|
workload = workloads.get(external_id)
|
|
if not workload:
|
|
workload = Workload(cluster_id=cluster.id, node_id=node.id, external_id=external_id, name=external_id, kind="qemu")
|
|
db.add(workload)
|
|
workloads[external_id] = workload
|
|
workload.node_id = node.id
|
|
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", []))
|
|
|
|
networks = {network.name: network for network in db.scalars(select(Network).where(Network.cluster_id == cluster.id)).all()}
|
|
for raw_network in inventory.get("networks", []):
|
|
name = raw_network.get("name") or raw_network.get("iface") or raw_network.get("id")
|
|
if not name:
|
|
continue
|
|
network = networks.get(name)
|
|
if not network:
|
|
network = Network(cluster_id=cluster.id, name=name, kind=raw_network.get("type") or "network")
|
|
db.add(network)
|
|
networks[name] = network
|
|
network.kind = raw_network.get("type") or raw_network.get("kind") or network.kind
|
|
vlan = raw_network.get("vlan") or raw_network.get("vlan_id")
|
|
network.vlan_id = int(vlan) if vlan not in (None, "") else network.vlan_id
|
|
|
|
db.add(Job(kind=job_kind, status="success", progress=100, logs=[f"Auto synced {cluster.name}"]))
|
|
db.commit()
|
|
return {"cluster": cluster.name, "status": "success", "inventory_counts": {key: len(value) for key, value in inventory.items()}}
|
|
|
|
|
|
async def discover_ipam(db: Session, job_kind: str = "ipam.auto_discover") -> dict[str, Any]:
|
|
imported = 0
|
|
removed = cleanup_discovered_container_networks(db)
|
|
errors = []
|
|
for cluster in db.scalars(select(Cluster).order_by(Cluster.name)).all():
|
|
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,
|
|
)
|
|
)
|
|
workloads = {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", []):
|
|
workload = workloads.get(str(raw_workload.get("vmid") or raw_workload.get("id") or ""))
|
|
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=job_kind,
|
|
status="success" if not errors else "failed",
|
|
progress=100,
|
|
logs=[f"Imported {imported} IP addresses", f"Removed {removed} container bridge IPs"],
|
|
error=str(errors) if errors else None,
|
|
)
|
|
)
|
|
db.commit()
|
|
return {"imported": imported, "removed": removed, "errors": errors}
|
|
|
|
|
|
async def run_due_jobs(db: Session) -> list[dict[str, Any]]:
|
|
setting = runtime_setting(db)
|
|
value = dict(setting.value or {})
|
|
results = []
|
|
if due(value, "auto_node_sync_enabled", "auto_node_sync_interval_minutes", "last_node_auto_sync_at"):
|
|
for cluster in db.scalars(select(Cluster).order_by(Cluster.name)).all():
|
|
results.append(await sync_cluster_inventory(db, cluster))
|
|
value["last_node_auto_sync_at"] = datetime.utcnow().isoformat()
|
|
if due(value, "auto_ipam_sync_enabled", "auto_ipam_sync_interval_minutes", "last_ipam_auto_sync_at"):
|
|
results.append(await discover_ipam(db))
|
|
value["last_ipam_auto_sync_at"] = datetime.utcnow().isoformat()
|
|
if results:
|
|
setting.value = value
|
|
db.commit()
|
|
return results
|