[DO-NOT-MERGE]Debug tags direction issue

.github/workflows/main.yml

@@ -36,7 +36,16 @@ on:
         options: 
           - legacy
           - ebpf
-        default: legacy  
+        default: legacy
+      Architecture:
+        description: "The target architecture(s) for the Docker image."
+        required: true
+        type: choice
+        options:
+          - both
+          - arm64
+          - amd64
+        default: both
 
 # A workflow run is made up of one or more jobs that can run sequentially or in parallel
 jobs:
@@ -82,11 +91,19 @@ jobs:
           ECR_REPOSITORY: akto-api-security
           REGISTRY_ALIAS: p7q3h0z2
           IMAGE_TAG: ${{ github.event.inputs.Tag }}
+          ARCH_INPUT: ${{ github.event.inputs.Architecture }}
         run: |
           # Build a docker container and push it to DockerHub 
           docker buildx create --use
-          echo "Building and Pushing image to ECR..."
-          docker buildx build --platform linux/arm64/v8,linux/amd64 -t $ECR_REGISTRY/$REGISTRY_ALIAS/mirror-api-logging:$IMAGE_TAG . --push
+          if [ "$ARCH_INPUT" == "arm64" ]; then
+            PLATFORM="linux/arm64/v8"
+          elif [ "$ARCH_INPUT" == "amd64" ]; then
+            PLATFORM="linux/amd64"
+          else
+            PLATFORM="linux/arm64/v8,linux/amd64"
+          fi
+          echo "Building and Pushing image to ECR with platform: $PLATFORM"
+          docker buildx build --platform $PLATFORM -t $ECR_REGISTRY/$REGISTRY_ALIAS/mirror-api-logging:$IMAGE_TAG . --push
           echo "::set-output name=image::$ECR_REGISTRY/$REGISTRY_ALIAS/mirror-api-logging:$IMAGE_TAG"
 
       - name: Build, tag, and push the image to Amazon ECR -ebpf
@@ -97,11 +114,19 @@ jobs:
           ECR_REPOSITORY: akto-api-security
           REGISTRY_ALIAS: p7q3h0z2
           IMAGE_TAG: ${{ github.event.inputs.EbpfTag }}
+          ARCH_INPUT: ${{ github.event.inputs.Architecture }}
         run: |
           # Build a docker container and push it to DockerHub 
           docker buildx create --use
-          echo "Building and Pushing image to ECR..."
-          docker buildx build --platform linux/arm64/v8,linux/amd64 -t $ECR_REGISTRY/$REGISTRY_ALIAS/mirror-api-logging:$IMAGE_TAG -f Dockerfile.eBPF . --push
+          if [ "$ARCH_INPUT" == "arm64" ]; then
+            PLATFORM="linux/arm64/v8"
+          elif [ "$ARCH_INPUT" == "amd64" ]; then
+            PLATFORM="linux/amd64"
+          else
+            PLATFORM="linux/arm64/v8,linux/amd64"
+          fi
+          echo "Building and Pushing image to ECR with platform: $PLATFORM"
+          docker buildx build --platform $PLATFORM -t $ECR_REGISTRY/$REGISTRY_ALIAS/mirror-api-logging:$IMAGE_TAG -f Dockerfile.eBPF . --push
           echo "::set-output name=image::$ECR_REGISTRY/$REGISTRY_ALIAS/mirror-api-logging:$IMAGE_TAG"
 
   build-docker:
@@ -136,11 +161,19 @@ jobs:
         env:
           ECR_REGISTRY: aktosecurity
           IMAGE_TAG: ${{ github.event.inputs.Tag }}
+          ARCH_INPUT: ${{ github.event.inputs.Architecture }}
         run: |
           # Build a docker container and push it to DockerHub 
           docker buildx create --use
-          echo "Building and Pushing image to DockerHub..."
-          docker buildx build --platform linux/arm64/v8,linux/amd64 -t $ECR_REGISTRY/mirror-api-logging:$IMAGE_TAG . --push
+          if [ "$ARCH_INPUT" == "arm64" ]; then
+            PLATFORM="linux/arm64/v8"
+          elif [ "$ARCH_INPUT" == "amd64" ]; then
+            PLATFORM="linux/amd64"
+          else
+            PLATFORM="linux/arm64/v8,linux/amd64"
+          fi
+          echo "Building and Pushing image to DockerHub with platform: $PLATFORM"
+          docker buildx build --platform $PLATFORM -t $ECR_REGISTRY/mirror-api-logging:$IMAGE_TAG . --push
           echo "::set-output name=image::$ECR_REGISTRY/mirror-api-logging:$IMAGE_TAG"
 
       - name: Build, tag, and push the image to DockerHub - ebpf
@@ -149,9 +182,17 @@ jobs:
         env:
           ECR_REGISTRY: aktosecurity
           IMAGE_TAG: ${{ github.event.inputs.EbpfTag }}
+          ARCH_INPUT: ${{ github.event.inputs.Architecture }}
         run: |
           # Build a docker container and push it to DockerHub 
           docker buildx create --use
-          echo "Building and Pushing image to DockerHub..."
-          docker buildx build --platform linux/arm64/v8,linux/amd64 -t $ECR_REGISTRY/mirror-api-logging:$IMAGE_TAG -f Dockerfile.eBPF . --push
+          if [ "$ARCH_INPUT" == "arm64" ]; then
+            PLATFORM="linux/arm64/v8"
+          elif [ "$ARCH_INPUT" == "amd64" ]; then
+            PLATFORM="linux/amd64"
+          else
+            PLATFORM="linux/arm64/v8,linux/amd64"
+          fi
+          echo "Building and Pushing image to DockerHub with platform: $PLATFORM"
+          docker buildx build --platform $PLATFORM -t $ECR_REGISTRY/mirror-api-logging:$IMAGE_TAG -f Dockerfile.eBPF . --push
           echo "::set-output name=image::$ECR_REGISTRY/mirror-api-logging:$IMAGE_TAG"

.gitignore

@@ -2,4 +2,5 @@ mirroring-api-logging
 .idea/
 **/.vscode/
 temp
-**temp
+**temp
+data-*

Dockerfile.eBPF

@@ -1,4 +1,4 @@
-FROM alpine:3.21 AS base
+FROM alpine:3.22 AS base
 
 USER root
 RUN apk add bcc-tools bcc-dev bcc-doc linux-headers build-base

ebpf-run.sh

@@ -2,7 +2,10 @@
 
 LOG_FILE="/tmp/dump.log"
 MAX_LOG_SIZE=${MAX_LOG_SIZE:-10485760}  # Default to 10 MB if not set (10 MB = 10 * 1024 * 1024 bytes)
-CHECK_INTERVAL=60                        # Check interval in seconds
+CHECK_INTERVAL=${CHECK_INTERVAL:-60}
+CHECK_INTERVAL_MEM=${CHECK_INTERVAL_MEM:-10}     # Check interval in seconds (configurable via env)
+MEMORY_THRESHOLD=${MEMORY_THRESHOLD:-80} # Kill process at this % memory usage (configurable via env)
+GOMEMLIMIT_PERCENT=${GOMEMLIMIT_PERCENT:-60} # GOMEMLIMIT as % of container memory limit (configurable via env)
 
 # Function to rotate the log file
 rotate_log() {
@@ -14,6 +17,30 @@ rotate_log() {
     fi
 }
 
+# Function to check memory usage and kill process if threshold exceeded
+check_memory_and_kill() {
+    # Get current memory usage in bytes
+    if [ -f /sys/fs/cgroup/memory.current ]; then
+        # cgroup v2
+        CURRENT_MEM=$(cat /sys/fs/cgroup/memory.current)
+    elif [ -f /sys/fs/cgroup/memory/memory.usage_in_bytes ]; then
+        # cgroup v1
+        CURRENT_MEM=$(cat /sys/fs/cgroup/memory/memory.usage_in_bytes)
+    else
+        return
+    fi
+
+    # Calculate percentage used
+    PERCENT_USED=$((CURRENT_MEM * 100 / MEM_LIMIT_BYTES))
+
+    echo "Memory usage: ${PERCENT_USED}% (${CURRENT_MEM} / ${MEM_LIMIT_BYTES} bytes)"
+
+    if [ "$PERCENT_USED" -ge "$MEMORY_THRESHOLD" ]; then
+        echo "Memory threshold ${MEMORY_THRESHOLD}% exceeded (${PERCENT_USED}%), killing ebpf-logging process"
+        pkill -9 ebpf-logging
+    fi
+}
+
 # Start monitoring in the background
 if [[ "${ENABLE_LOGS}" == "false" ]]; then
     while true; do
@@ -22,10 +49,52 @@ if [[ "${ENABLE_LOGS}" == "false" ]]; then
     done &
 fi
 
+# 1. Check if MEM_LIMIT is provided as env variable
+if [ -z "$MEM_LIMIT" ]; then
+    # Not provided, detect and read cgroup memory limits
+    if [ -f /sys/fs/cgroup/memory.max ]; then
+        # cgroup v2
+        MEM_LIMIT_BYTES=$(cat /sys/fs/cgroup/memory.max)
+    elif [ -f /sys/fs/cgroup/memory/memory.limit_in_bytes ]; then
+        # cgroup v1
+        MEM_LIMIT_BYTES=$(cat /sys/fs/cgroup/memory/memory.limit_in_bytes)
+    else
+        # Fallback to free -b (bytes) if cgroup file not found
+        echo "Neither cgroup v2 nor v1 memory file found, defaulting to free -m"
+        # Convert from kB to bytes
+        MEM_LIMIT_BYTES=$(free -b | awk '/Mem:/ {print $2}')
+    fi
+
+    # 2. Handle edge cases: "max" means no strict limit or a very large limit
+    if [ "$MEM_LIMIT_BYTES" = "max" ]; then
+        # Arbitrary fallback (1 GiB in bytes here, but adjust as needed)
+        echo "Cgroup memory limit set to 'max', defaulting to free memory"
+        MEM_LIMIT_BYTES=$(free -b | awk '/Mem:/ {print $2}')
+    fi
+
+    # 3. Convert the memory limit from bytes to MB (integer division)
+    MEM_LIMIT_MB=$((MEM_LIMIT_BYTES / 1024 / 1024))
+else
+    # MEM_LIMIT provided as env variable, treat as MB
+    echo "Using MEM_LIMIT from environment variable: ${MEM_LIMIT} MB"
+    MEM_LIMIT_MB=$MEM_LIMIT
+    # Convert MB to bytes for calculations
+    MEM_LIMIT_BYTES=$((MEM_LIMIT * 1024 * 1024))
+fi
+
+echo "Using container memory limit: ${MEM_LIMIT_MB} MB"
+
+# Set GOMEMLIMIT for the Go process
+GOMEMLIMIT_MB=$((MEM_LIMIT_MB * GOMEMLIMIT_PERCENT / 100))
+export GOMEMLIMIT="${GOMEMLIMIT_MB}MiB"
+echo "Setting GOMEMLIMIT to: ${GOMEMLIMIT} (${GOMEMLIMIT_PERCENT}% of ${MEM_LIMIT_MB} MB)"
+
+# Start memory monitoring in the background
+
 while :
 do
 if [[ "${ENABLE_LOGS}" == "false" ]]; then
-    ./ebpf-logging >> "$LOG_FILE" 2>&1 
+    ./ebpf-logging >> "$LOG_FILE" 2>&1
 else
     ./ebpf-logging
 fi

ebpf/bpfwrapper/eventCallbacks.go

@@ -178,6 +178,7 @@ func SocketDataEventCallback(inputChan chan []byte, connectionFactory *connectio
 			"data", dataStr,
 			"rc", event.Attr.ReadEventsCount,
 			"wc", event.Attr.WriteEventsCount,
-			"ssl", event.Attr.Ssl)
+			"ssl", event.Attr.Ssl,
+			"bytesSent", bytesSent)
 	}
 }

ebpf/connections/factory.go

@@ -73,13 +73,15 @@ func convertToSingleByteArr(bufMap map[int][]byte) []byte {
 var (
 	disableEgress        = false
 	maxActiveConnections = 4096
-	inactivityThreshold  = 3 * time.Second
+	inactivityThreshold  = 7 * time.Second
 	// Value in MB
 	bufferMemThreshold = 400
 
 	// unique id of daemonset
 	uniqueDaemonsetId          = uuid.New().String()
 	trackerDataProcessInterval = 100
+
+	socketDataEventBytesThreshold = 10 * 1024 * 1024
 )
 
 func init() {
@@ -89,6 +91,7 @@ func init() {
 	utils.InitVar("TRAFFIC_BUFFER_THRESHOLD", &bufferMemThreshold)
 	utils.InitVar("AKTO_MEM_SOFT_LIMIT", &bufferMemThreshold)
 	utils.InitVar("TRACKER_DATA_PROCESS_INTERVAL", &trackerDataProcessInterval)
+	utils.InitVar("SOCKET_DATA_EVENT_BYTES_THRESHOLD", &socketDataEventBytesThreshold)
 }
 
 func ProcessTrackerData(connID structs.ConnID, tracker *Tracker, isComplete bool) {
@@ -201,6 +204,25 @@ func (factory *Factory) CreateIfNotExists(connectionID structs.ConnID) {
 	}
 }
 
+// resetTimer stops, drains, and resets the timer to the given duration.
+func resetTimer(t *time.Timer, d time.Duration) {
+	if !t.Stop() {
+		select {
+		case <-t.C:
+		default:
+		}
+	}
+	t.Reset(d)
+}
+
+// Worker lifecycle:
+//   ACTIVE:
+//     - socket data/open -> reset inactivity timer on each event
+//     - socket close     -> schedule delayed termination
+//     - inactivity timer -> terminate immediately
+//
+//   TERMINATION is final and happens exactly once.
+//  either due to inactivityThreshold or due to socker close event
 func (factory *Factory) StartWorker(connectionID structs.ConnID, tracker *Tracker, ch chan interface{}) {
 	go func(connID structs.ConnID, tracker *Tracker, ch chan interface{}) {
 
@@ -216,9 +238,17 @@ func (factory *Factory) StartWorker(connectionID structs.ConnID, tracker *Tracke
 				case *structs.SocketDataEvent:
 					utils.LogProcessing("Received data event", "fd", connID.Fd, "id", connID.Id, "timestamp", connID.Conn_start_ns, "ip", connID.Ip, "port", connID.Port)
 					tracker.AddDataEvent(*e)
+					if tracker.GetSentBytes() + tracker.GetRecvBytes() > uint64(socketDataEventBytesThreshold) {
+						utils.LogProcessing("Socket Data threshold data breached, processing current data", "fd", connID.Fd, "id", connID.Id, "timestamp", connID.Conn_start_ns, "ip", connID.Ip, "port", connID.Port)
+						factory.StopProcessing(connID)
+						return
+					}else{
+						resetTimer(inactivityTimer, inactivityThreshold)
+					}
 				case *structs.SocketOpenEvent:
 					utils.LogProcessing("Received open event", "fd", connID.Fd, "id", connID.Id, "timestamp", connID.Conn_start_ns, "ip", connID.Ip, "port", connID.Port)
 					tracker.AddOpenEvent(*e)
+					resetTimer(inactivityTimer, inactivityThreshold)
 				case *structs.SocketCloseEvent:
 					utils.LogProcessing("Received close event", "fd", connID.Fd, "id", connID.Id, "timestamp", connID.Conn_start_ns, "ip", connID.Ip, "port", connID.Port)
 					tracker.AddCloseEvent(*e)
@@ -230,22 +260,25 @@ func (factory *Factory) StartWorker(connectionID structs.ConnID, tracker *Tracke
 
 			case <-delayedDeleteChan:
 				utils.LogProcessing("Stopping go routine (delayed close)", "fd", connID.Fd, "id", connID.Id, "timestamp", connID.Conn_start_ns, "ip", connID.Ip, "port", connID.Port)
-				factory.ProcessAndStopWorker(connID)
-				factory.DeleteWorker(connID)
+				factory.StopProcessing(connID)
 				return
 
 			case <-inactivityTimer.C:
 				// Eat the go routine after inactive threshold, process the tracker and stop the worker
 				utils.LogProcessing("Inactivity threshold reached, marking connection as inactive and processing", "fd", connID.Fd, "id", connID.Id, "timestamp", connID.Conn_start_ns, "ip", connID.Ip, "port", connID.Port)
-				factory.ProcessAndStopWorker(connID)
-				factory.DeleteWorker(connID)
+				factory.StopProcessing(connID)
 				utils.LogProcessing("Stopping go routine", "fd", connID.Fd, "id", connID.Id, "timestamp", connID.Conn_start_ns, "ip", connID.Ip, "port", connID.Port)
 				return
 			}
 		}
 	}(connectionID, tracker, ch)
 }
 
+func (factory *Factory) StopProcessing(connID structs.ConnID){
+	factory.ProcessAndStopWorker(connID)
+	factory.DeleteWorker(connID)
+}
+
 func (factory *Factory) ProcessAndStopWorker(connectionID structs.ConnID) {
 	tracker, connExists := factory.getTracker(connectionID)
 	if connExists {

ebpf/connections/parser.go

@@ -5,5 +5,18 @@ import (
 )
 
 func tryReadFromBD(ip string, destIp string, receiveBuffer []byte, sentBuffer []byte, isComplete bool, direction int, id uint64, fd uint32, daemonsetIdentifier, hostName string) {
-	kafkaUtil.ParseAndProduce(receiveBuffer, sentBuffer, ip, destIp, 0, false, "MIRRORING", isComplete, direction, id, fd, daemonsetIdentifier, hostName)
+	ctx := kafkaUtil.TrafficContext{
+		SourceIP:            ip,
+		DestIP:              destIp,
+		VxlanID:             0,
+		IsPending:           false,
+		TrafficSource:       "MIRRORING",
+		IsComplete:          isComplete,
+		Direction:           direction,
+		ProcessID:           uint32(id >> 32),
+		SocketFD:            fd,
+		DaemonsetIdentifier: daemonsetIdentifier,
+		HostName:            hostName,
+	}
+	kafkaUtil.ParseAndProduce(receiveBuffer, sentBuffer, ctx)
 }

ebpf/connections/tracker.go

@@ -108,3 +108,11 @@ func (conn *Tracker) AddCloseEvent(event structs.SocketCloseEvent) {
 	conn.closeTimestamp = uint64(time.Now().UnixNano())
 	conn.lastAccessTimestamp = uint64(time.Now().UnixNano())
 }
+
+func (conn *Tracker) GetSentBytes() uint64 {
+	return conn.sentBytes
+}
+
+func (conn *Tracker) GetRecvBytes() uint64 {
+	return conn.recvBytes
+}