feat: add automatic cluster sync and IPAM discovery with configurable intervals, firewall log file parsing, and enhanced flow prioritization

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

backend/app/agent_assets/nexafabric-agent.py

@@ -18,7 +18,7 @@ from pathlib import Path
 from typing import Any
 
 
-VERSION = "0.2.1"
+VERSION = "0.2.2"
 VM_INTERFACE_RE = re.compile(r"(?:tap|fwbr|fwln|fwpr)(\d+)")
 VM_INTERFACE_DETAIL_RE = re.compile(r"(?:tap|fwbr|fwln|fwpr)(\d+)i(\d+)")
 LOG_FIELD_RE = re.compile(r"\b([A-Z]+)=([^\s]+)")
@@ -362,18 +362,39 @@ def parse_firewall_log_line(line: str) -> dict[str, Any] | None:
     }
 
 
+def firewall_log_lines_from_files(max_lines: int = 5000) -> tuple[list[str], list[dict[str, str]]]:
+    lines: list[str] = []
+    errors: list[dict[str, str]] = []
+    for path in ("/var/log/pve-firewall.log", "/var/log/pve-firewall.log.1"):
+        try:
+            with open(path, "r", encoding="utf-8", errors="ignore") as handle:
+                file_lines = handle.readlines()[-max_lines:]
+                lines.extend(line.rstrip("\n") for line in file_lines)
+        except OSError as exc:
+            errors.append({"path": path, "error": str(exc)})
+    return lines[-max_lines:], errors
+
+
 def collect_firewall_log_flows(since_minutes: int = 5, limit: int = 500) -> tuple[list[dict[str, Any]], dict[str, Any]]:
-    diagnostics = {"collector": "journalctl-kernel", "since_minutes": since_minutes, "errors": []}
+    diagnostics = {"collector": "journalctl-kernel+pve-firewall-log", "since_minutes": since_minutes, "errors": []}
     code, output = run_command(
         ["journalctl", "-k", "--since", f"-{max(since_minutes, 1)} min", "--no-pager", "-o", "cat"],
         timeout=10,
     )
-    if code != 0 or not output:
+    log_lines: list[str] = []
+    if code == 0 and output:
+        log_lines.extend(output.splitlines())
+    else:
         diagnostics["errors"].append(output or "journalctl returned no firewall log output")
+    file_lines, file_errors = firewall_log_lines_from_files()
+    log_lines.extend(file_lines)
+    diagnostics["file_errors"] = file_errors
+    diagnostics["lines_scanned"] = len(log_lines)
+    if not log_lines:
         return [], diagnostics
 
     flows: dict[tuple[object, ...], dict[str, Any]] = {}
-    for line in output.splitlines():
+    for line in log_lines:
         if "SRC=" not in line or "DST=" not in line:
             continue
         flow = parse_firewall_log_line(line)
@@ -418,7 +439,16 @@ def merge_flow_sources(*sources: list[dict[str, Any]], limit: int = 500) -> list
                 current["bytes"] = int(current.get("bytes") or 0) + int(flow.get("bytes") or 0)
                 continue
             flows[key] = dict(flow)
-    return sorted(flows.values(), key=lambda item: int(item.get("bytes") or 0), reverse=True)[:limit]
+    return sorted(
+        flows.values(),
+        key=lambda item: (
+            1 if str(item.get("decision") or item.get("state") or "").lower() in {"blocked", "drop", "dropped", "reject", "rejected", "deny", "denied"} else 0,
+            1 if item.get("collector") == "firewall-log" else 0,
+            int(item.get("bytes") or 0),
+            int(item.get("packets") or 0),
+        ),
+        reverse=True,
+    )[:limit]
 
 
 def collect_conntrack() -> dict[str, Any]:
@@ -470,7 +500,7 @@ def collect_firewall() -> dict[str, Any]:
 def collect_payload(config: dict[str, Any]) -> dict[str, Any]:
     uptime = read_text("/proc/uptime")
     interfaces = collect_interfaces()
-    flow_limit = int(config.get("flow_limit", 500))
+    flow_limit = int(config.get("flow_limit", 2000))
     packet_flows: list[dict[str, Any]] = []
     packet_diagnostics: dict[str, Any] | None = None
     firewall_log_flows: list[dict[str, Any]] = []

backend/app/api/v1/router.py

@@ -107,7 +107,16 @@ def setup_setting(db: Session) -> SystemSetting:
 def runtime_setting(db: Session) -> SystemSetting:
     setting = db.get(SystemSetting, "runtime")
     if not setting:
-        setting = SystemSetting(key="runtime", value={"flow_retention_hours": 24})
+        setting = SystemSetting(
+            key="runtime",
+            value={
+                "flow_retention_hours": 24,
+                "auto_node_sync_enabled": False,
+                "auto_node_sync_interval_minutes": 60,
+                "auto_ipam_sync_enabled": False,
+                "auto_ipam_sync_interval_minutes": 60,
+            },
+        )
         db.add(setting)
         db.commit()
         db.refresh(setting)
@@ -116,7 +125,15 @@ def runtime_setting(db: Session) -> SystemSetting:
 
 def runtime_settings_payload(db: Session) -> RuntimeSettingsRead:
     value = runtime_setting(db).value or {}
-    return RuntimeSettingsRead(flow_retention_hours=int(value.get("flow_retention_hours") or 24))
+    return RuntimeSettingsRead(
+        flow_retention_hours=int(value.get("flow_retention_hours") or 24),
+        auto_node_sync_enabled=bool(value.get("auto_node_sync_enabled", False)),
+        auto_node_sync_interval_minutes=int(value.get("auto_node_sync_interval_minutes") or 60),
+        auto_ipam_sync_enabled=bool(value.get("auto_ipam_sync_enabled", False)),
+        auto_ipam_sync_interval_minutes=int(value.get("auto_ipam_sync_interval_minutes") or 60),
+        last_node_auto_sync_at=datetime.fromisoformat(value["last_node_auto_sync_at"]) if value.get("last_node_auto_sync_at") else None,
+        last_ipam_auto_sync_at=datetime.fromisoformat(value["last_ipam_auto_sync_at"]) if value.get("last_ipam_auto_sync_at") else None,
+    )
 
 
 def require_super_admin(user: User) -> None:
@@ -1313,7 +1330,7 @@ cat > "$CONFIG_DIR/config.json" <<'JSON'
   "node_id": "{node.id}",
   "node_name": "{node.name}",
   "interval_seconds": 30,
-  "flow_limit": 500,
+  "flow_limit": 2000,
   "packet_flow_collector": true,
   "packet_flow_window_seconds": 10,
   "firewall_log_collector": true,
@@ -1448,7 +1465,7 @@ def agent_heartbeat(payload: AgentHeartbeat, authorization: str | None = Header(
         ): flow
         for flow in db.scalars(select(TrafficFlow).where(TrafficFlow.node_id == node.id)).all()
     }
-    for raw_flow in payload.flows[:1000]:
+    for raw_flow in payload.flows[:5000]:
         source_ip = str(raw_flow.get("source_ip") or "")
         destination_ip = str(raw_flow.get("destination_ip") or "")
         if not source_ip or not destination_ip:
@@ -1517,7 +1534,7 @@ async def workload_insights(workload_id: str, _: CurrentUser, db: Session = Depe
             select(TrafficFlow)
             .where((TrafficFlow.source_ip.in_(workload_ips)) | (TrafficFlow.destination_ip.in_(workload_ips)))
             .order_by((TrafficFlow.state == "blocked").desc(), TrafficFlow.updated_at.desc())
-            .limit(50)
+            .limit(1000)
         ).all()
         for flow in flows:
             source_owner = ip_owners.get(flow.source_ip)
@@ -2068,7 +2085,8 @@ def update_settings(payload: RuntimeSettingsUpdate, user: CurrentUser, db: Sessi
     require_super_admin(user)
     setting = runtime_setting(db)
     old_values = dict(setting.value or {})
-    setting.value = {**old_values, "flow_retention_hours": payload.flow_retention_hours}
+    changes = payload.model_dump(exclude_unset=True, exclude_none=True)
+    setting.value = {**old_values, **changes}
     commit_or_400(db)
     write_audit(
         db,

backend/app/schemas/domain.py

@@ -173,10 +173,20 @@ class RuntimeSettingsRead(BaseModel):
     firewall_apply_requires_preview: bool = True
     agent_optional: bool = True
     flow_retention_hours: int = 24
+    auto_node_sync_enabled: bool = False
+    auto_node_sync_interval_minutes: int = 60
+    auto_ipam_sync_enabled: bool = False
+    auto_ipam_sync_interval_minutes: int = 60
+    last_node_auto_sync_at: datetime | None = None
+    last_ipam_auto_sync_at: datetime | None = None
 
 
 class RuntimeSettingsUpdate(BaseModel):
-    flow_retention_hours: int = Field(ge=1, le=8760)
+    flow_retention_hours: int | None = Field(default=None, ge=1, le=8760)
+    auto_node_sync_enabled: bool | None = None
+    auto_node_sync_interval_minutes: int | None = Field(default=None, ge=1, le=10080)
+    auto_ipam_sync_enabled: bool | None = None
+    auto_ipam_sync_interval_minutes: int | None = Field(default=None, ge=1, le=10080)
 
 
 class ClusterRead(OrmModel):

backend/app/services/auto_sync.py

@@ -0,0 +1,235 @@
+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

backend/app/workers/worker.py

@@ -1,12 +1,25 @@
-import time
+import asyncio
+
+from app.db.session import SessionLocal
+from app.services.auto_sync import run_due_jobs
+
+
+async def loop() -> None:
+    print("NexaFabric worker started. Auto-sync scheduler is active.", flush=True)
+    while True:
+        try:
+            with SessionLocal() as db:
+                results = await run_due_jobs(db)
+                for result in results:
+                    print(f"auto-sync: {result}", flush=True)
+        except Exception as exc:
+            print(f"auto-sync failed: {exc}", flush=True)
+        await asyncio.sleep(30)
 
 
 def main() -> None:
-    print("NexaFabric worker started. Configure Celery queues for production job execution.", flush=True)
-    while True:
-        time.sleep(30)
+    asyncio.run(loop())
 
 
 if __name__ == "__main__":
     main()
-

backend/tests/test_agent_firewall_logs.py

@@ -0,0 +1,30 @@
+import importlib.util
+from pathlib import Path
+
+
+def load_agent_module():
+    path = Path(__file__).resolve().parents[1] / "app" / "agent_assets" / "nexafabric-agent.py"
+    spec = importlib.util.spec_from_file_location("nexafabric_agent", path)
+    assert spec and spec.loader
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+def test_parse_proxmox_reject_firewall_log_line():
+    agent = load_agent_module()
+    line = (
+        "100 6 tap100i0-IN 09/Jul/2026:23:04:58 +0200 REJECT: IN=fwbr100i0 OUT=fwbr100i0 "
+        "PHYSIN=fwln100i0 PHYSOUT=tap100i0 SRC=172.16.155.74 DST=172.16.0.100 LEN=52 "
+        "TTL=128 ID=47107 PROTO=TCP SPT=58945 DPT=80"
+    )
+
+    flow = agent.parse_firewall_log_line(line)
+
+    assert flow["source_ip"] == "172.16.155.74"
+    assert flow["destination_ip"] == "172.16.0.100"
+    assert flow["protocol"] == "tcp"
+    assert flow["source_port"] == 58945
+    assert flow["destination_port"] == 80
+    assert flow["decision"] == "blocked"
+    assert flow["state"] == "blocked"