Merge pull request #108 from akto-api-security/feat/tags-only

Revert "Merge pull request #105 from akto-api-security/event_loss_fixes"

Author: gauravakto

ebpf/bpfwrapper/perfbufferreaders.go

@@ -2,7 +2,6 @@ package bpfwrapper
 
 import (
 	"fmt"
-	"github.com/akto-api-security/mirroring-api-logging/trafficUtil/kafkaUtil"
 	"log"
 
 	"github.com/iovisor/gobpf/bcc"
@@ -37,7 +36,7 @@ func NewProbeChannel(name string, handler ProbeEventLoop) *ProbeChannel {
 
 // Start initiate a goroutine for the event loop handler, for a lost events messages and the perf map.
 func (probeChannel *ProbeChannel) Start(module *bcc.Module, connectionFactory *connections.Factory) error {
-	probeChannel.eventChannel = make(chan []byte, kafkaUtil.EventChanBuffSize)
+	probeChannel.eventChannel = make(chan []byte)
 	probeChannel.lostEventsChannel = make(chan uint64)
 
 	table := bcc.NewTable(module.TableId(probeChannel.name), module)

ebpf/connections/factory.go

@@ -1,7 +1,6 @@
 package connections
 
 import (
-	"bytes"
 	"encoding/binary"
 	"fmt"
 	"log/slog"
@@ -16,86 +15,19 @@ import (
 	"github.com/google/uuid"
 )
 
-var httpBytes = []byte("HTTP")
-
 // Factory is a routine-safe container that holds a trackers with unique ID, and able to create new tracker.
 type Factory struct {
-	processor        map[structs.ConnID]chan interface{}
-	connections      map[structs.ConnID]*Tracker
-	mutex            *sync.RWMutex
-	trackersToDelete sync.Map
+	processor   map[structs.ConnID]chan interface{}
+	connections map[structs.ConnID]*Tracker
+	mutex       *sync.RWMutex
 }
 
 // NewFactory creates a new instance of the factory.
 func NewFactory() *Factory {
-	f := &Factory{
-		processor:        make(map[structs.ConnID]chan interface{}),
-		connections:      make(map[structs.ConnID]*Tracker),
-		mutex:            &sync.RWMutex{},
-		trackersToDelete: sync.Map{},
-	}
-
-	f.StartCleanupWorker()
-	return f
-}
-
-func (factory *Factory) StartCleanupWorker() {
-	go func() {
-		for {
-			time.Sleep(100 * time.Millisecond)
-
-			now := time.Now()
-			factory.trackersToDelete.Range(func(key, value interface{}) bool {
-				connID := key.(structs.ConnID)
-				markedAt := value.(time.Time)
-
-				if now.Sub(markedAt) > time.Duration(trackerDataProcessInterval)*time.Millisecond {
-					factory.ProcessAndStopWorker(connID)
-					factory.forceDeleteWorker(connID)
-					factory.trackersToDelete.Delete(connID)
-				}
-				return true
-			})
-		}
-	}()
-}
-
-func (factory *Factory) forceDeleteWorker(connectionID structs.ConnID) {
-	factory.mutex.Lock()
-	defer factory.mutex.Unlock()
-
-	if ch, exists := factory.processor[connectionID]; exists {
-		close(ch)
-		delete(factory.processor, connectionID)
-		slog.Debug("Deleted event channel", "fd", connectionID.Fd, "id", connectionID.Id, "timestamp", connectionID.Conn_start_ns, "ip", connectionID.Ip, "port", connectionID.Port)
-	}
-
-	if _, exists := factory.connections[connectionID]; exists {
-		delete(factory.connections, connectionID)
-		slog.Debug("Deleted connection", "fd", connectionID.Fd, "id", connectionID.Id, "timestamp", connectionID.Conn_start_ns, "ip", connectionID.Ip, "port", connectionID.Port)
-		requestProcessCount++
-	}
-
-	if (time.Now().UnixMilli())-lastMemCheck > int64(memCheckInterval) {
-		lastMemCheck = time.Now().UnixMilli()
-		mem := utils.LogMemoryStats()
-
-		utils.PrintLog("Requests processed", "count", requestProcessCount, "lastMemCheck", lastMemCheck)
-		utils.PrintLog("connection factory size", "connections", len(factory.connections), "processors", len(factory.processor), "lastMemCheck", lastMemCheck)
-		requestProcessCount = 0
-		if mem >= bufferMemThreshold {
-			trackersToDelete := make(map[structs.ConnID]struct{})
-			for k := range factory.connections {
-				trackersToDelete[k] = struct{}{}
-			}
-			for key := range trackersToDelete {
-				if ch, exists := factory.processor[key]; exists {
-					close(ch)
-					delete(factory.processor, key)
-				}
-				delete(factory.connections, key)
-			}
-		}
+	return &Factory{
+		processor:   make(map[structs.ConnID]chan interface{}),
+		connections: make(map[structs.ConnID]*Tracker),
+		mutex:       &sync.RWMutex{},
 	}
 }
 
@@ -143,8 +75,7 @@ var (
 	bufferMemThreshold = 400
 
 	// unique id of daemonset
-	uniqueDaemonsetId          = uuid.New().String()
-	trackerDataProcessInterval = 100
+	uniqueDaemonsetId = uuid.New().String()
 )
 
 func init() {
@@ -153,7 +84,6 @@ func init() {
 	utils.InitVar("TRAFFIC_INACTIVITY_THRESHOLD", &inactivityThreshold)
 	utils.InitVar("TRAFFIC_BUFFER_THRESHOLD", &bufferMemThreshold)
 	utils.InitVar("AKTO_MEM_SOFT_LIMIT", &bufferMemThreshold)
-	utils.InitVar("TRACKER_DATA_PROCESS_INTERVAL", &trackerDataProcessInterval)
 }
 
 func ProcessTrackerData(connID structs.ConnID, tracker *Tracker, isComplete bool) {
@@ -185,14 +115,10 @@ func ProcessTrackerData(connID structs.ConnID, tracker *Tracker, isComplete bool
 		hostName = kafkaUtil.PodInformerInstance.GetPodNameByProcessId(int32(connID.Id>>32))
 	}
 
-	if len(sentBuffer) >= len(httpBytes) && (bytes.Equal(sentBuffer[:len(httpBytes)], httpBytes)) {
-		tryReadFromBD(destIpStr, srcIpStr, receiveBuffer, sentBuffer, isComplete, 1, connID.Id, connID.Fd, uniqueDaemonsetId, hostName)
-	}
+	tryReadFromBD(destIpStr, srcIpStr, receiveBuffer, sentBuffer, isComplete, 1, connID.Id, connID.Fd, uniqueDaemonsetId, hostName)
 	if !disableEgress {
 		// attempt to parse the egress as well by switching the recv and sent buffers.
-		if len(receiveBuffer) >= len(httpBytes) && (bytes.Equal(receiveBuffer[:len(httpBytes)], httpBytes)) {
-			tryReadFromBD(srcIpStr, destIpStr, sentBuffer, receiveBuffer, isComplete, 2, connID.Id, connID.Fd, uniqueDaemonsetId, hostName)
-		}
+		tryReadFromBD(srcIpStr, destIpStr, sentBuffer, receiveBuffer, isComplete, 2, connID.Id, connID.Fd, uniqueDaemonsetId, hostName)
 	}
 }
 
@@ -286,13 +212,16 @@ func (factory *Factory) StartWorker(connectionID structs.ConnID, tracker *Tracke
 				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)
+					factory.ProcessAndStopWorker(connID)
 					factory.DeleteWorker(connID)
 					utils.LogProcessing("Stopping go routine", "fd", connID.Fd, "id", connID.Id, "timestamp", connID.Conn_start_ns, "ip", connID.Ip, "port", connID.Port)
+					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)
 				utils.LogProcessing("Stopping go routine", "fd", connID.Fd, "id", connID.Id, "timestamp", connID.Conn_start_ns, "ip", connID.Ip, "port", connID.Port)
 				return
@@ -312,7 +241,40 @@ func (factory *Factory) ProcessAndStopWorker(connectionID structs.ConnID) {
 func (factory *Factory) DeleteWorker(connectionID structs.ConnID) {
 	factory.mutex.Lock()
 	defer factory.mutex.Unlock()
-	factory.trackersToDelete.Store(connectionID, time.Now())
+
+	if ch, exists := factory.processor[connectionID]; exists {
+		close(ch)
+		delete(factory.processor, connectionID)
+		utils.LogProcessing("Deleted event channel", "fd", connectionID.Fd, "id", connectionID.Id, "timestamp", connectionID.Conn_start_ns, "ip", connectionID.Ip, "port", connectionID.Port)
+	}
+
+	if _, exists := factory.connections[connectionID]; exists {
+		delete(factory.connections, connectionID)
+		utils.LogProcessing("Deleted connection", "fd", connectionID.Fd, "id", connectionID.Id, "timestamp", connectionID.Conn_start_ns, "ip", connectionID.Ip, "port", connectionID.Port)
+		requestProcessCount++
+	}
+
+	if (time.Now().UnixMilli())-lastMemCheck > int64(memCheckInterval) {
+		lastMemCheck = time.Now().UnixMilli()
+		mem := utils.LogMemoryStats()
+		utils.PrintLog("Requests processed", "count", requestProcessCount, "lastMemCheck", lastMemCheck)
+		utils.PrintLog("connection factory size", "connections", len(factory.connections), "processors", len(factory.processor), "lastMemCheck", lastMemCheck)
+		requestProcessCount = 0
+		if mem >= bufferMemThreshold {
+			trackersToDelete := make(map[structs.ConnID]struct{})
+			utils.LogProcessing("Deleting all trackers at mem", "mem", mem)
+			for k := range factory.connections {
+				trackersToDelete[k] = struct{}{}
+			}
+			for key := range trackersToDelete {
+				if ch, exists := factory.processor[key]; exists {
+					close(ch)
+					delete(factory.processor, key)
+				}
+				delete(factory.connections, key)
+			}
+		}
+	}
 }
 
 func (factory *Factory) getChannel(connectionID structs.ConnID) (chan interface{}, bool) {

trafficUtil/kafkaUtil/kafka.go

@@ -306,7 +306,7 @@ func getKafkaWriter(kafkaURL string, batchSize int, batchTimeout time.Duration)
 		WriteTimeout: batchTimeout,
 		Async:        true,
 		Balancer:     &kafka.Hash{},
-		Compression:  kafka.Zstd,
+		Compression:  kafka.Lz4,
 	}
 
 	if useTLS {

trafficUtil/kafkaUtil/parser.go

@@ -40,19 +40,14 @@ var (
 		"TRACE":   true,
 		"TRACK":   true,
 		"PATCH":   true}
-	DebugStrings     = []string{}
-	globalReader     = &bytes.Reader{}
-	globalReaderLock sync.Mutex
-
-	EventChanBuffSize = 100000
+	DebugStrings = []string{}
 )
 
 const ONE_MINUTE = 60
 
 func init() {
 	utils.InitVar("DEBUG_MODE", &debugMode)
 	utils.InitVar("OUTPUT_BANDWIDTH_LIMIT", &outputBandwidthLimitPerMin)
-	utils.InitVar("EVENT_CHAN_BUFF_SIZE", &EventChanBuffSize)
 	// convert MB to B
 	if outputBandwidthLimitPerMin != -1 {
 		outputBandwidthLimitPerMin = outputBandwidthLimitPerMin * 1024 * 1024
@@ -66,6 +61,10 @@ func init() {
 
 	// Start ticker to read debug URLs from file every 30 seconds
 	go func() {
+		if !utils.FileLoggingEnabled {
+			slog.Info("File logging is not enabled, skipping debug URL file watcher")
+			return
+		}
 		ticker := time.NewTicker(30 * time.Second)
 		defer ticker.Stop()
 		for {
@@ -131,7 +130,7 @@ func checkDebugUrlAndPrint(url string, host string, message string) {
 				utils.PrintLogDebug(logMsg)
 				go ProduceLogs(ctx, logMsg, LogTypeInfo)
 				break
-			}else if strings.Contains(host, debugString) {
+			} else if strings.Contains(host, debugString) {
 				ctx := context.Background()
 				logMsg := fmt.Sprintf("%s : %s", message, host)
 				utils.PrintLogDebug(logMsg)
@@ -140,7 +139,6 @@ func checkDebugUrlAndPrint(url string, host string, message string) {
 			}
 		}
 	}
-	slog.Debug("EventChanBuffSize value ", EventChanBuffSize)
 }
 
 func checkAndUpdateBandwidthProcessed(sampleSize int) bool {
@@ -463,7 +461,7 @@ func ParseAndProduce(receiveBuffer []byte, sentBuffer []byte, sourceIp string, d
 				}
 			}
 		} else {
-			checkDebugUrlAndPrint(url,req.Host, "Pod labels not resolved, PodInformerInstance is nil or direction is not inbound, direction: "+fmt.Sprint(direction))
+			checkDebugUrlAndPrint(url, req.Host, "Pod labels not resolved, PodInformerInstance is nil or direction is not inbound, direction: "+fmt.Sprint(direction))
 		}
 
 		out, _ := json.Marshal(value)
@@ -509,4 +507,4 @@ func ParseAndProduce(receiveBuffer []byte, sentBuffer []byte, sourceIp string, d
 
 		i++
 	}
-}
+}

trafficUtil/utils/logger.go

@@ -14,13 +14,15 @@ var (
 	processLogs  bool = false
 	aktoLogLevel string
 	level        slog.Level    = slog.LevelWarn
+	FileLoggingEnabled bool = false
 )
 
 func SetupLogger() {
 	slog.Warn("Setting up logger")
 	InitVar("INGEST_LOGS", &ingestLogs)
 	InitVar("PROCESS_LOGS", &processLogs)
 	InitVar("AKTO_LOG_LEVEL", &aktoLogLevel)
+	InitVar("AKTO_FILE_LOGGING_ENABLED", &FileLoggingEnabled)
 
 	if aktoLogLevel != "" {
 		switch strings.ToUpper(aktoLogLevel) {
@@ -73,10 +75,8 @@ const HOST_MAPPING_PATH = "/ebpf/logs/akto/"
 const LOG_ROTATE_INTERVAL = 60 * 5 // 5 minutes
 
 const (
-	OSPidLogFile         = HOST_MAPPING_PATH + "ospidlog.txt"
 	GoPidLogFile         = HOST_MAPPING_PATH + "gopidlog.txt"
 	LabelsMapLogFile     = HOST_MAPPING_PATH + "labelsmaplog.txt"
-	ResolveLabelsLogFile = HOST_MAPPING_PATH + "resolvelabels.txt"
 )
 
 func SetupFileLogger(filePath string) {
@@ -93,6 +93,11 @@ func SetupFileLogger(filePath string) {
 }
 
 func LogToSpecificFile(filePath string, message string, args ...any) {
+	if !FileLoggingEnabled {
+		slog.Warn("File logging is disabled, skipping log to file", "filePath", filePath)
+		return
+	}
+
 	textLogger, exists := fileHandlers[filePath]
 	if !exists {
 		slog.Error("Logger not initialized for", "filePath", filePath)
@@ -114,13 +119,15 @@ func LogToSpecificFile(filePath string, message string, args ...any) {
 }
 
 func SetupAllFileLoggers() {
+	if !FileLoggingEnabled {
+		slog.Warn("File logging is disabled, skipping setup")
+		return
+	}
 	if err := os.MkdirAll(HOST_MAPPING_PATH, 0755); err != nil {
 		slog.Error("Failed to create log directory", "path", HOST_MAPPING_PATH, "error", err)
 	}
-	SetupFileLogger(OSPidLogFile)
 	SetupFileLogger(GoPidLogFile)
 	SetupFileLogger(LabelsMapLogFile)
-	SetupFileLogger(ResolveLabelsLogFile)
 }
 
 // TODO: Call this somewhere in the shutdown process