feat: add timestamp tracking to flows with first_seen_at/last_seen_at/observed_at fields across all collectors

Add FIREWALL_LOG_TS_RE regex to parse timestamps from firewall log lines, implement firewall_log_seen_at to extract and convert log timestamps to UTC ISO format, add first_seen_at/last_seen_at/observed_at fields to flows in collect_packet_flows (AF_PACKET collector) with timestamp updates on flow aggregation, add timestamp fields to collect_flows (conntrack collector) and parse_firewall_log_line (

backend/app/agent_assets/nexafabric-agent.py

@@ -22,6 +22,7 @@ VERSION = "0.3.0"
 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]+)")
+FIREWALL_LOG_TS_RE = re.compile(r"\b(\d{2}/[A-Za-z]{3}/\d{4}:\d{2}:\d{2}:\d{2} [+-]\d{4})\b")
 ETH_P_IP = 0x0800
 ETH_P_ALL = 0x0003
 ETH_P_8021Q = 0x8100
@@ -255,13 +256,18 @@ def collect_packet_flows(interfaces: list[dict[str, Any]], duration: int, limit:
                 if current:
                     current["packets"] += 1
                     current["bytes"] += int(flow["bytes"])
+                    current["last_seen_at"] = datetime.now(timezone.utc).isoformat()
                     continue
+                now = datetime.now(timezone.utc).isoformat()
                 flow.update(
                     {
                         "vmid": vmid,
                         "nic": nic,
                         "interface": name,
                         "collector": "linux-af-packet",
+                        "first_seen_at": now,
+                        "last_seen_at": now,
+                        "observed_at": now,
                     }
                 )
                 flows[key] = flow
@@ -332,12 +338,27 @@ def collect_flows(limit: int = 500) -> list[dict[str, Any]]:
         if key in seen:
             continue
         seen.add(key)
+        now = datetime.now(timezone.utc).isoformat()
+        flow["first_seen_at"] = now
+        flow["last_seen_at"] = now
+        flow["observed_at"] = now
+        flow["collector"] = flow.get("collector") or "conntrack"
         flows.append(flow)
         if len(flows) >= limit:
             break
     return flows
 
 
+def firewall_log_seen_at(line: str) -> str:
+    match = FIREWALL_LOG_TS_RE.search(line)
+    if match:
+        try:
+            return datetime.strptime(match.group(1), "%d/%b/%Y:%H:%M:%S %z").astimezone(timezone.utc).isoformat()
+        except ValueError:
+            pass
+    return datetime.now(timezone.utc).isoformat()
+
+
 def parse_firewall_log_line(line: str) -> dict[str, Any] | None:
     fields = {key: value for key, value in LOG_FIELD_RE.findall(line)}
     source_ip = fields.get("SRC")
@@ -358,6 +379,7 @@ def parse_firewall_log_line(line: str) -> dict[str, Any] | None:
     source_port = fields.get("SPT")
     destination_port = fields.get("DPT")
     length = fields.get("LEN")
+    seen_at = firewall_log_seen_at(line)
     return {
         "source_ip": source_ip,
         "destination_ip": destination_ip,
@@ -370,6 +392,9 @@ def parse_firewall_log_line(line: str) -> dict[str, Any] | None:
         "decision": decision,
         "collector": "firewall-log",
         "log_excerpt": line[-180:],
+        "first_seen_at": seen_at,
+        "last_seen_at": seen_at,
+        "observed_at": seen_at,
     }
 
 
@@ -423,6 +448,8 @@ def collect_firewall_log_flows(since_minutes: int = 5, limit: int = 500) -> tupl
         if current:
             current["packets"] += 1
             current["bytes"] += int(flow.get("bytes") or 0)
+            current["last_seen_at"] = flow.get("last_seen_at") or datetime.now(timezone.utc).isoformat()
+            current["observed_at"] = current["last_seen_at"]
             continue
         flows[key] = flow
         if len(flows) >= limit:
@@ -448,6 +475,11 @@ def merge_flow_sources(*sources: list[dict[str, Any]], limit: int = 500) -> list
             if current:
                 current["packets"] = int(current.get("packets") or 0) + int(flow.get("packets") or 0)
                 current["bytes"] = int(current.get("bytes") or 0) + int(flow.get("bytes") or 0)
+                current_seen = str(current.get("last_seen_at") or current.get("observed_at") or "")
+                flow_seen = str(flow.get("last_seen_at") or flow.get("observed_at") or "")
+                if flow_seen > current_seen:
+                    current["last_seen_at"] = flow_seen
+                    current["observed_at"] = flow_seen
                 continue
             flows[key] = dict(flow)
     return sorted(

backend/app/agent_assets/nexafabric-ebpf.go

@@ -55,13 +55,16 @@ type flowValue struct {
 	Direction       string `json:"direction,omitempty"`
 	State           string `json:"state"`
 	Collector       string `json:"collector"`
+	FirstSeenAt     string `json:"first_seen_at"`
+	LastSeenAt      string `json:"last_seen_at"`
+	ObservedAt      string `json:"observed_at"`
 }
 
 type diagnostics struct {
-	AttachMode         string   `json:"attach_mode"`
+	AttachMode          string   `json:"attach_mode"`
 	InterfacesRequested []string `json:"interfaces_requested"`
 	InterfacesAttached  []string `json:"interfaces_attached"`
-	Errors             []string `json:"errors"`
+	Errors              []string `json:"errors"`
 }
 
 type payload struct {
@@ -189,9 +192,11 @@ func openSocket(interfaceName string) (int, error) {
 
 func collect(interfaceNames []string, duration time.Duration, limit int) payload {
 	result := payload{
+		Flows: []flowValue{},
 		Diagnostics: diagnostics{
 			AttachMode:          "af_packet_raw_socket",
 			InterfacesRequested: interfaceNames,
+			InterfacesAttached:  []string{},
 			Errors:              []string{},
 		},
 	}
@@ -258,6 +263,7 @@ func collect(interfaceNames []string, duration time.Duration, limit int) payload
 			if !ok {
 				continue
 			}
+			now := time.Now().UTC().Format(time.RFC3339Nano)
 			key.Interface = socket.name
 			key.VMID, key.NIC = interfaceMeta(socket.name)
 			key.Direction = "ingress"
@@ -278,11 +284,16 @@ func collect(interfaceNames []string, duration time.Duration, limit int) payload
 					Direction:       key.Direction,
 					State:           "observed",
 					Collector:       "ebpf-helper",
+					FirstSeenAt:     now,
+					LastSeenAt:      now,
+					ObservedAt:      now,
 				}
 				flows[key] = current
 			}
 			current.Packets++
 			current.Bytes += uint64(bytes)
+			current.LastSeenAt = now
+			current.ObservedAt = now
 			if len(flows) >= limit {
 				break
 			}

backend/app/api/v1/router.py

@@ -656,6 +656,20 @@ def traffic_flow_key(node_id: str, raw_flow: dict[str, object]) -> tuple[object,
     )
 
 
+def flow_observed_at(raw_flow: dict[str, object], payload_collected_at: datetime | None) -> datetime:
+    for key in ("last_seen_at", "observed_at", "first_seen_at"):
+        value = raw_flow.get(key)
+        if not value:
+            continue
+        try:
+            parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
+        except ValueError:
+            continue
+        return parsed.replace(tzinfo=None) if parsed.tzinfo else parsed
+    fallback = payload_collected_at or datetime.utcnow()
+    return fallback.replace(tzinfo=None) if fallback.tzinfo else fallback
+
+
 def compact_agent_payload(payload: AgentHeartbeat) -> dict:
     value = payload.model_dump(mode="json")
     flows = value.get("flows")
@@ -1581,8 +1595,7 @@ def agent_heartbeat(payload: AgentHeartbeat, authorization: str | None = Header(
         destination_ip = str(raw_flow.get("destination_ip") or "")
         if not source_ip or not destination_ip:
             continue
-        observed_at = payload.collected_at or datetime.utcnow()
+        observed_at = flow_observed_at(raw_flow, payload.collected_at)
-        observed_at = observed_at.replace(tzinfo=None) if observed_at.tzinfo else observed_at
         key = traffic_flow_key(node.id, raw_flow)
         existing = existing_flows.get(key)
         if existing:

frontend/src/pages/Workloads.tsx

@@ -94,6 +94,12 @@ function summarizeTraffic(traffic: Array<Record<string, unknown>>) {
       }
       if (!existing.observedAt && flow.observed_at) {
         existing.observedAt = String(flow.observed_at);
+      } else if (flow.observed_at) {
+        const existingTime = Date.parse(existing.observedAt || "");
+        const flowTime = Date.parse(String(flow.observed_at));
+        if (Number.isFinite(flowTime) && (!Number.isFinite(existingTime) || flowTime > existingTime)) {
+          existing.observedAt = String(flow.observed_at);
+        }
       }
       continue;
     }