Merge pull request #25 from hengyoush/feature/netanalysis

[Improvement] Clean kern event stream when connection parser make progress

.gitmodules

@@ -1,6 +1,6 @@
 [submodule "libbpf"]
 	path = libbpf
-	url = https://mirror.ghproxy.com/github.com/libbpf/libbpf.git
+	url = https://github.com/libbpf/libbpf.git
 [submodule "bpftool"]
 	path = bpftool
-	url = https://mirror.ghproxy.com/github.com/libbpf/bpftool
+	url = https://github.com/libbpf/bpftool

agent/analysis/stat.go

@@ -233,6 +233,8 @@ func (s *StatRecorder) ReceiveRecord(r protocol.Record, connection *conn.Connect
 		annotatedRecord.reqNicEventDetails = KernEventsToEventDetails[NicEventDetail](devOutSyscallEvents)
 		annotatedRecord.respNicEventDetails = KernEventsToEventDetails[NicEventDetail](nicIngressEvents)
 	}
+	streamEvents.DiscardEventsBySeq(egressMessage.Seq()+uint64(egressMessage.ByteSize()), true)
+	streamEvents.DiscardEventsBySeq(ingressMessage.Seq()+uint64(ingressMessage.ByteSize()), false)
 	if recordsChannel == nil {
 		log.Infoln(annotatedRecord.String(AnnotatedRecordToStringOptions{
 			Nano: false,

agent/conn/kern_event_handler.go

@@ -83,6 +83,24 @@ func (s *KernEventStream) FindAndRemoveEventsBySeqAndLen(step bpf.AgentStepT, se
 	return result
 }
 
+func (s *KernEventStream) DiscardEventsBySeq(seq uint64, egress bool) {
+	for step, events := range s.kernEvents {
+		if egress && !bpf.IsEgressStep(step) {
+			continue
+		}
+		if !egress && !bpf.IsIngressStep(step) {
+			continue
+		}
+		index, _ := slices.BinarySearchFunc(events, KernEvent{seq: seq}, func(i KernEvent, j KernEvent) int {
+			return cmp.Compare(i.seq, j.seq)
+		})
+		discardIdx := index
+		if discardIdx > 0 {
+			s.kernEvents[step] = events[discardIdx:]
+		}
+	}
+}
+
 type KernEvent struct {
 	seq       uint64
 	len       int

bpf/types.go

@@ -9,3 +9,11 @@ type SyscallEventData struct {
 	SyscallEvent SyscallEvent
 	Buf          []byte
 }
+
+func IsEgressStep(step AgentStepT) bool {
+	return step <= AgentStepTNIC_OUT
+}
+
+func IsIngressStep(step AgentStepT) bool {
+	return step >= AgentStepTNIC_IN
+}