Merge pull request #93 from ddosify/develop

Develop

aggregator/data.go

@@ -295,11 +295,11 @@ func (a *Aggregator) Run() {
 	for i := 0; i < numWorker; i++ {
 		go a.processEbpf(a.ctx)
 		go a.processEbpfTcp(a.ctx)
+		go a.processEbpfProc(a.ctx)
 	}
 
 	for i := 0; i < 2*cpuCount; i++ {
 		go a.processHttp2Frames()
-		go a.processEbpfProc(a.ctx)
 	}
 }
 
@@ -1127,8 +1127,6 @@ func (a *Aggregator) fetchSkInfo(ctx context.Context, skLine *SocketLine, d *l7_
 	var skInfo *SockInfo
 	var err error
 
-	// skInfo, _ = skLine.GetValue(d.WriteTimeNs)
-
 	for {
 		skInfo, err = skLine.GetValue(d.WriteTimeNs)
 		if err == nil && skInfo != nil {
@@ -1172,18 +1170,31 @@ func (a *Aggregator) fetchSocketMap(pid uint32) *SocketMap {
 }
 
 func (a *Aggregator) findRelatedSocket(ctx context.Context, d *l7_req.L7Event) *SockInfo {
-	sockMap := a.fetchSocketMap(d.Pid)
-	skLine := a.fetchSkLine(sockMap, d.Pid, d.Fd)
-	skInfo := a.fetchSkInfo(ctx, skLine, d)
+	sockMap := a.clusterInfo.SocketMaps[d.Pid]
+	// acquire sockMap lock
+	sockMap.mu.Lock()
+
+	if sockMap.M == nil {
+		sockMap.M = make(map[uint64]*SocketLine)
+	}
 
+	skLine, ok := sockMap.M[d.Fd]
+	if !ok {
+		log.Logger.Debug().Uint32("pid", d.Pid).Uint64("fd", d.Fd).Msg("error finding skLine, go look for it")
+		// start new socket line, find already established connections
+		skLine = NewSocketLine(d.Pid, d.Fd)
+		sockMap.M[d.Fd] = skLine
+	}
+
+	// release sockMap lock
+	sockMap.mu.Unlock()
+
+	skInfo := a.fetchSkInfo(ctx, skLine, d)
 	if skInfo == nil {
 		return nil
 	}
 
-	// TODO: zero IP address check ??
-
 	return skInfo
-
 }
 
 func (a *Aggregator) parseSqlCommand(d *l7_req.L7Event) (string, error) {

main_benchmark_test.go

@@ -11,6 +11,7 @@ import (
 	"os"
 	"runtime/metrics"
 	"runtime/pprof"
+	"strconv"
 
 	"sync"
 	"sync/atomic"
@@ -383,20 +384,25 @@ func (s *Simulator) Setup() {
 	// Create Kubernetes Workloads
 	// K8sResourceMessage
 
+	maxPid, err := getPidMax()
+	if err != nil {
+		simLog.Fatal().Err(err).Msg("could not get pid max")
+	}
+
 	for i := 0; i < s.conf.PodCount; i++ {
 		// TODO: namespace
 		podName := fake.Name()
 		podIP := fake.IP(fake.WithIPv4())
 		mainContainerImage := fake.Name()
 		uid := types.UID(fake.Name())
-		pid := rand.Uint32()
+		pid := rand.Int31n(int32(maxPid))
 
 		s.pods[podName] = &FakePod{
 			Name:            podName,
 			IP:              podIP,
 			Image:           mainContainerImage,
 			Uid:             uid,
-			Pid:             pid,
+			Pid:             uint32(pid),
 			Fds:             map[uint64]struct{}{},
 			OpenConnections: map[uint64]uint64{},
 		}
@@ -696,3 +702,25 @@ func ToText(number uint32) string {
 	return fmt.Sprintf("%d", number)
 
 }
+
+func getPidMax() (int, error) {
+	// Read the contents of the file
+	f, err := os.Open("/proc/sys/kernel/pid_max")
+	if err != nil {
+		fmt.Println("Error opening file:", err)
+		return 0, err
+	}
+	content, err := io.ReadAll(f)
+	if err != nil {
+		fmt.Println("Error reading file:", err)
+		return 0, err
+	}
+
+	// Convert the content to an integer
+	pidMax, err := strconv.Atoi(string(content[:len(content)-1])) // trim newline
+	if err != nil {
+		fmt.Println("Error converting to integer:", err)
+		return 0, err
+	}
+	return pidMax, nil
+}