feat: add interface traffic counters as fallback telemetry when conntrack flows unavailable
Add interface_traffic collection in agent to aggregate VM/LXC network counters by vmid/nic with tap/fwln/fwpr/fwbr interface ranking, implement collect_interface_traffic to select best interface per VM NIC and format as flow-like records with rx/tx bytes/packets, add collect_flow_diagnostics to capture conntrack binary path and kernel bridge/netfilter settings for debugging, update workload_insights endpoint
This commit is contained in:
@@ -18,6 +18,7 @@ from typing import Any
|
|||||||
|
|
||||||
VERSION = "0.1.0"
|
VERSION = "0.1.0"
|
||||||
VM_INTERFACE_RE = re.compile(r"(?:tap|fwbr|fwln|fwpr)(\d+)")
|
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+)")
|
||||||
|
|
||||||
|
|
||||||
def read_text(path: str) -> str | None:
|
def read_text(path: str) -> str | None:
|
||||||
@@ -30,7 +31,8 @@ def read_text(path: str) -> str | None:
|
|||||||
def run_command(args: list[str], timeout: int = 5) -> tuple[int, str]:
|
def run_command(args: list[str], timeout: int = 5) -> tuple[int, str]:
|
||||||
try:
|
try:
|
||||||
result = subprocess.run(args, capture_output=True, text=True, timeout=timeout, check=False)
|
result = subprocess.run(args, capture_output=True, text=True, timeout=timeout, check=False)
|
||||||
return result.returncode, result.stdout.strip()
|
output = result.stdout.strip() or result.stderr.strip()
|
||||||
|
return result.returncode, output
|
||||||
except (OSError, subprocess.SubprocessError):
|
except (OSError, subprocess.SubprocessError):
|
||||||
return 127, ""
|
return 127, ""
|
||||||
|
|
||||||
@@ -56,6 +58,58 @@ def collect_interfaces() -> list[dict[str, Any]]:
|
|||||||
return interfaces
|
return interfaces
|
||||||
|
|
||||||
|
|
||||||
|
def interface_rank(name: str) -> int:
|
||||||
|
if name.startswith("tap"):
|
||||||
|
return 0
|
||||||
|
if name.startswith("fwln"):
|
||||||
|
return 1
|
||||||
|
if name.startswith("fwpr"):
|
||||||
|
return 2
|
||||||
|
if name.startswith("fwbr"):
|
||||||
|
return 3
|
||||||
|
return 9
|
||||||
|
|
||||||
|
|
||||||
|
def collect_interface_traffic(interfaces: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||||||
|
selected: dict[tuple[str, str], dict[str, Any]] = {}
|
||||||
|
for interface in interfaces:
|
||||||
|
vmid = interface.get("vmid")
|
||||||
|
if not vmid:
|
||||||
|
continue
|
||||||
|
detail_match = VM_INTERFACE_DETAIL_RE.search(str(interface.get("name") or ""))
|
||||||
|
nic = detail_match.group(2) if detail_match else "0"
|
||||||
|
key = (str(vmid), nic)
|
||||||
|
current = selected.get(key)
|
||||||
|
if current and interface_rank(str(current.get("name") or "")) <= interface_rank(str(interface.get("name") or "")):
|
||||||
|
continue
|
||||||
|
selected[key] = interface
|
||||||
|
|
||||||
|
traffic = []
|
||||||
|
for (vmid, nic), interface in sorted(selected.items()):
|
||||||
|
rx_bytes = int(interface.get("rx_bytes") or 0)
|
||||||
|
tx_bytes = int(interface.get("tx_bytes") or 0)
|
||||||
|
rx_packets = int(interface.get("rx_packets") or 0)
|
||||||
|
tx_packets = int(interface.get("tx_packets") or 0)
|
||||||
|
traffic.append(
|
||||||
|
{
|
||||||
|
"vmid": vmid,
|
||||||
|
"nic": nic,
|
||||||
|
"interface": interface.get("name"),
|
||||||
|
"source": f"vm:{vmid}",
|
||||||
|
"destination": "network",
|
||||||
|
"protocol": "interface-counter",
|
||||||
|
"rx_bytes": rx_bytes,
|
||||||
|
"tx_bytes": tx_bytes,
|
||||||
|
"bytes": rx_bytes + tx_bytes,
|
||||||
|
"rx_packets": rx_packets,
|
||||||
|
"tx_packets": tx_packets,
|
||||||
|
"packets": rx_packets + tx_packets,
|
||||||
|
"state": interface.get("operstate") or "unknown",
|
||||||
|
}
|
||||||
|
)
|
||||||
|
return traffic
|
||||||
|
|
||||||
|
|
||||||
def parse_conntrack_line(line: str) -> dict[str, Any] | None:
|
def parse_conntrack_line(line: str) -> dict[str, Any] | None:
|
||||||
parts = line.split()
|
parts = line.split()
|
||||||
if len(parts) < 5 or parts[0] not in {"tcp", "udp", "icmp"}:
|
if len(parts) < 5 or parts[0] not in {"tcp", "udp", "icmp"}:
|
||||||
@@ -132,6 +186,20 @@ def collect_conntrack() -> dict[str, Any]:
|
|||||||
return {"count": None, "source": "unavailable"}
|
return {"count": None, "source": "unavailable"}
|
||||||
|
|
||||||
|
|
||||||
|
def collect_flow_diagnostics(conntrack: dict[str, Any], flow_count: int) -> dict[str, Any]:
|
||||||
|
code, conntrack_path = run_command(["sh", "-c", "command -v conntrack"])
|
||||||
|
return {
|
||||||
|
"flow_count": flow_count,
|
||||||
|
"conntrack_binary": conntrack_path if code == 0 else None,
|
||||||
|
"conntrack_count": conntrack.get("count"),
|
||||||
|
"conntrack_source": conntrack.get("source"),
|
||||||
|
"bridge_nf_call_iptables": read_text("/proc/sys/net/bridge/bridge-nf-call-iptables"),
|
||||||
|
"bridge_nf_call_ip6tables": read_text("/proc/sys/net/bridge/bridge-nf-call-ip6tables"),
|
||||||
|
"nf_conntrack_max": read_text("/proc/sys/net/netfilter/nf_conntrack_max"),
|
||||||
|
"nf_conntrack_count": read_text("/proc/sys/net/netfilter/nf_conntrack_count"),
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
def collect_firewall() -> dict[str, Any]:
|
def collect_firewall() -> dict[str, Any]:
|
||||||
status = {}
|
status = {}
|
||||||
code, output = run_command(["systemctl", "is-active", "pve-firewall"])
|
code, output = run_command(["systemctl", "is-active", "pve-firewall"])
|
||||||
@@ -146,6 +214,9 @@ def collect_firewall() -> dict[str, Any]:
|
|||||||
|
|
||||||
def collect_payload(config: dict[str, Any]) -> dict[str, Any]:
|
def collect_payload(config: dict[str, Any]) -> dict[str, Any]:
|
||||||
uptime = read_text("/proc/uptime")
|
uptime = read_text("/proc/uptime")
|
||||||
|
interfaces = collect_interfaces()
|
||||||
|
flows = collect_flows(int(config.get("flow_limit", 500)))
|
||||||
|
conntrack = collect_conntrack()
|
||||||
return {
|
return {
|
||||||
"version": VERSION,
|
"version": VERSION,
|
||||||
"node_name": config.get("node_name"),
|
"node_name": config.get("node_name"),
|
||||||
@@ -154,11 +225,12 @@ def collect_payload(config: dict[str, Any]) -> dict[str, Any]:
|
|||||||
"kernel": platform.release(),
|
"kernel": platform.release(),
|
||||||
"uptime_seconds": float(uptime.split()[0]) if uptime else None,
|
"uptime_seconds": float(uptime.split()[0]) if uptime else None,
|
||||||
"loadavg": list(os.getloadavg()) if hasattr(os, "getloadavg") else [],
|
"loadavg": list(os.getloadavg()) if hasattr(os, "getloadavg") else [],
|
||||||
"interfaces": collect_interfaces(),
|
"interfaces": interfaces,
|
||||||
"flows": collect_flows(int(config.get("flow_limit", 500))),
|
"interface_traffic": collect_interface_traffic(interfaces),
|
||||||
"conntrack": collect_conntrack(),
|
"flows": flows,
|
||||||
|
"conntrack": conntrack,
|
||||||
"firewall": collect_firewall(),
|
"firewall": collect_firewall(),
|
||||||
"extra": {"platform": platform.platform()},
|
"extra": {"platform": platform.platform(), "flow_diagnostics": collect_flow_diagnostics(conntrack, len(flows))},
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -894,6 +894,32 @@ def workload_insights(workload_id: str, _: CurrentUser, db: Session = Depends(ge
|
|||||||
"ip_addresses": [flow.source_ip, flow.destination_ip],
|
"ip_addresses": [flow.source_ip, flow.destination_ip],
|
||||||
}
|
}
|
||||||
)
|
)
|
||||||
|
if not traffic:
|
||||||
|
agent = db.get(NodeAgent, workload.node_id)
|
||||||
|
payload = agent.last_payload if agent and isinstance(agent.last_payload, dict) else {}
|
||||||
|
for item in payload.get("interface_traffic", []):
|
||||||
|
if not isinstance(item, dict) or str(item.get("vmid")) != str(workload.external_id):
|
||||||
|
continue
|
||||||
|
traffic.append(
|
||||||
|
{
|
||||||
|
"source": workload.name,
|
||||||
|
"destination": "network",
|
||||||
|
"interface": item.get("interface"),
|
||||||
|
"protocol": item.get("protocol") or "interface-counter",
|
||||||
|
"port": None,
|
||||||
|
"bytes": item.get("bytes") or 0,
|
||||||
|
"packets": item.get("packets") or 0,
|
||||||
|
"rx_bytes": item.get("rx_bytes") or 0,
|
||||||
|
"tx_bytes": item.get("tx_bytes") or 0,
|
||||||
|
"rx_packets": item.get("rx_packets") or 0,
|
||||||
|
"tx_packets": item.get("tx_packets") or 0,
|
||||||
|
"state": item.get("state") or "unknown",
|
||||||
|
"decision": "observed",
|
||||||
|
"observed_at": payload.get("collected_at"),
|
||||||
|
"ip_addresses": workload_ips,
|
||||||
|
"note": "Interface counter fallback. No host conntrack flows were available.",
|
||||||
|
}
|
||||||
|
)
|
||||||
audit_mode_notes = [
|
audit_mode_notes = [
|
||||||
f"{policy.name} is in audit mode; matching traffic is logged without enforcement."
|
f"{policy.name} is in audit mode; matching traffic is logged without enforcement."
|
||||||
for policy in policies
|
for policy in policies
|
||||||
|
|||||||
@@ -203,6 +203,7 @@ class AgentHeartbeat(BaseModel):
|
|||||||
uptime_seconds: float | None = None
|
uptime_seconds: float | None = None
|
||||||
loadavg: list[float] = []
|
loadavg: list[float] = []
|
||||||
interfaces: list[dict[str, Any]] = []
|
interfaces: list[dict[str, Any]] = []
|
||||||
|
interface_traffic: list[dict[str, Any]] = []
|
||||||
flows: list[dict[str, Any]] = []
|
flows: list[dict[str, Any]] = []
|
||||||
conntrack: dict[str, Any] = {}
|
conntrack: dict[str, Any] = {}
|
||||||
firewall: dict[str, Any] = {}
|
firewall: dict[str, Any] = {}
|
||||||
|
|||||||
@@ -32,7 +32,11 @@ export function Nodes() {
|
|||||||
|
|
||||||
function agentFlows(node: Node) {
|
function agentFlows(node: Node) {
|
||||||
const flows = node.agent?.last_payload?.flows;
|
const flows = node.agent?.last_payload?.flows;
|
||||||
return Array.isArray(flows) ? flows : [];
|
const interfaceTraffic = node.agent?.last_payload?.interface_traffic;
|
||||||
|
if (Array.isArray(flows) && flows.length) {
|
||||||
|
return flows;
|
||||||
|
}
|
||||||
|
return Array.isArray(interfaceTraffic) ? interfaceTraffic : [];
|
||||||
}
|
}
|
||||||
|
|
||||||
function agentInterfaces(node: Node) {
|
function agentInterfaces(node: Node) {
|
||||||
@@ -166,13 +170,21 @@ export function Nodes() {
|
|||||||
<div className="max-h-48 overflow-auto rounded-md border border-border">
|
<div className="max-h-48 overflow-auto rounded-md border border-border">
|
||||||
{agentFlows(detailNode).length ? agentFlows(detailNode).slice(0, 50).map((item, index) => {
|
{agentFlows(detailNode).length ? agentFlows(detailNode).slice(0, 50).map((item, index) => {
|
||||||
const flow = item as Record<string, unknown>;
|
const flow = item as Record<string, unknown>;
|
||||||
|
const isInterfaceCounter = flow.protocol === "interface-counter";
|
||||||
return (
|
return (
|
||||||
<div key={index} className="border-b border-border px-3 py-2 text-xs last:border-0">
|
<div key={index} className="border-b border-border px-3 py-2 text-xs last:border-0">
|
||||||
<div className="font-medium">{String(flow.source_ip)}:{String(flow.source_port ?? "")} {"->"} {String(flow.destination_ip)}:{String(flow.destination_port ?? "")}</div>
|
<div className="font-medium">
|
||||||
<div className="text-slate-500">{String(flow.protocol ?? "unknown")} · {String(flow.bytes ?? 0)} bytes · {String(flow.packets ?? 0)} packets · {String(flow.state ?? "unknown")}</div>
|
{isInterfaceCounter
|
||||||
|
? `VMID ${String(flow.vmid)} ${String(flow.interface ?? "")}`
|
||||||
|
: `${String(flow.source_ip)}:${String(flow.source_port ?? "")} -> ${String(flow.destination_ip)}:${String(flow.destination_port ?? "")}`}
|
||||||
|
</div>
|
||||||
|
<div className="text-slate-500">
|
||||||
|
{String(flow.protocol ?? "unknown")} · {String(flow.bytes ?? 0)} bytes · {String(flow.packets ?? 0)} packets · {String(flow.state ?? "unknown")}
|
||||||
|
</div>
|
||||||
|
{isInterfaceCounter ? <div className="text-slate-500">rx {String(flow.rx_bytes ?? 0)} · tx {String(flow.tx_bytes ?? 0)}</div> : null}
|
||||||
</div>
|
</div>
|
||||||
);
|
);
|
||||||
}) : <div className="p-3 text-xs text-slate-500">No conntrack flows were reported in the last heartbeat.</div>}
|
}) : <div className="p-3 text-xs text-slate-500">No conntrack flows or interface counters were reported in the last heartbeat.</div>}
|
||||||
</div>
|
</div>
|
||||||
</section>
|
</section>
|
||||||
<section>
|
<section>
|
||||||
|
|||||||
@@ -63,7 +63,9 @@ export function Workloads() {
|
|||||||
<div key={index} className="rounded-md border border-border p-3">
|
<div key={index} className="rounded-md border border-border p-3">
|
||||||
<div>{String(flow.source)} → {String(flow.destination)}</div>
|
<div>{String(flow.source)} → {String(flow.destination)}</div>
|
||||||
<div className="text-xs text-slate-500">{String(flow.protocol)}:{String(flow.port)} · {String(flow.bytes)} bytes · {String(flow.decision)}</div>
|
<div className="text-xs text-slate-500">{String(flow.protocol)}:{String(flow.port)} · {String(flow.bytes)} bytes · {String(flow.decision)}</div>
|
||||||
|
{flow.protocol === "interface-counter" ? <div className="mt-1 text-xs text-slate-500">Interface: {String(flow.interface ?? "unknown")} · rx {String(flow.rx_bytes ?? 0)} · tx {String(flow.tx_bytes ?? 0)}</div> : null}
|
||||||
{Array.isArray(flow.ip_addresses) ? <div className="mt-1 text-xs text-slate-500">IPs: {flow.ip_addresses.join(", ")}</div> : null}
|
{Array.isArray(flow.ip_addresses) ? <div className="mt-1 text-xs text-slate-500">IPs: {flow.ip_addresses.join(", ")}</div> : null}
|
||||||
|
{flow.note ? <div className="mt-1 text-xs text-slate-500">{String(flow.note)}</div> : null}
|
||||||
</div>
|
</div>
|
||||||
)) : <div className="rounded-md border border-border p-3 text-xs text-slate-500">No real traffic telemetry has been collected yet. Install the node agent or enable a flow source to populate this section.</div>}
|
)) : <div className="rounded-md border border-border p-3 text-xs text-slate-500">No real traffic telemetry has been collected yet. Install the node agent or enable a flow source to populate this section.</div>}
|
||||||
</div>
|
</div>
|
||||||
|
|||||||
Reference in New Issue
Block a user