Add logs on kafka push

Author: gauravakto

trafficUtil/kafkaUtil/kafka.go

@@ -86,7 +86,7 @@ func InitKafka() {
 
 		out, _ := json.Marshal(value)
 		ctx := context.Background()
-		err := ProduceStr(ctx, string(out))
+		err := ProduceStr(ctx, string(out), "testKafkaConnection", "testKafkaConnectionHost")
 		utils.PrintLog("logging kafka stats post pushing message")
 		LogKafkaStats()
 		if err != nil {
@@ -274,7 +274,7 @@ func ProduceLogs(ctx context.Context, message string, logType string) error {
 	return nil
 }
 
-func ProduceStr(ctx context.Context, message string) error {
+func ProduceStr(ctx context.Context, message string, url, reqHost string ) error {
 	// initialize the writer with the broker addresses, and the topic
 	topic := "akto.api.logs"
 	msg := kafka.Message{
@@ -288,6 +288,8 @@ func ProduceStr(ctx context.Context, message string) error {
 		slog.Error("ERROR while writing messages", "topic", topic, "error", err)
 		return err
 	}
+	checkDebugUrlAndPrint(url, reqHost, "Kafka write successful: " + message)
+
 	return nil
 }
 

trafficUtil/kafkaUtil/parser.go

@@ -504,7 +504,7 @@ func ParseAndProduce(receiveBuffer []byte, sentBuffer []byte, sourceIp string, d
 		} else {
 			// Produce to kafka
 			// TODO : remove and use protobuf instead
-			go ProduceStr(ctx, string(out))
+			go ProduceStr(ctx, string(out), url, req.Host)
 			go Produce(ctx, payload)
 		}
 

trafficUtil/kafkaUtil/podinformer.go

@@ -355,6 +355,7 @@ func (w *PodInformer) handlePodAdd(obj interface{}) {
 	slog.Debug("Pod added:", "namespace", pod.Namespace, "podName", pod.Name)
 	w.podNameLabelsMap.Store(pod.Name, pod.Labels)
 	// Build the PID to Hostname map again to ensure it is up-to-date
+	// TODO: Optimize this ? What's the rate of pod add events?
 	w.BuildPidHostNameMap()
 	go ProducePodMapping(context.Background(), pod.Name)
 }
@@ -384,5 +385,6 @@ func (w *PodInformer) handlePodDelete(obj interface{}) {
 	slog.Debug("Pod deleted:", "namespace", pod.Namespace, "podName", pod.Name)
 	w.podNameLabelsMap.Delete(pod.Name)
 	// Build the PID to Hostname map again to ensure it is up-to-date
+	// TODO: Optimize this ? What's the rate of pod add events?
 	w.BuildPidHostNameMap()
 }

trafficUtil/utils/logger.go

@@ -77,8 +77,14 @@ func HasLogIntervalPassed() bool {
 	return true
 }
 
-var fileHandlers = make(map[string]*os.File)
+type textLogger struct {
+	handler      *os.File
+	lastWriteTime int64
+}
+
+var fileHandlers = make(map[string]*textLogger)
 const HOST_MAPPING_PATH = "/ebpf/logs/akto/"
+const LOG_ROTATE_INTERVAL = 60 * 5 // 5 minutes
 
 const (
 	OSPidLogFile         = HOST_MAPPING_PATH + "ospidlog.txt"
@@ -89,7 +95,10 @@ const (
 
 func SetupFileLogger(filePath string) {
 	file, err := os.OpenFile(filePath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
-	fileHandlers[filePath] = file
+	fileHandlers[filePath] = &textLogger{
+		handler:      file,
+		lastWriteTime: time.Now().Unix(),
+	}
 	if err != nil {
 		slog.Error("Failed to open log file", "filePath", filePath, "error", err)
 		return
@@ -99,16 +108,24 @@ func SetupFileLogger(filePath string) {
 }
 
 func LogToSpecificFile(filePath string, message string, args ...any) {
-	handler, exists := fileHandlers[filePath]
+	textLogger, exists := fileHandlers[filePath]
 	if !exists {
 		slog.Error("Logger not initialized for", "filePath", filePath)
 		return
 	}
-	if _, err := handler.WriteString(message); err != nil {
+	now := time.Now().Unix()
+	if now-textLogger.lastWriteTime > LOG_ROTATE_INTERVAL {
+		// empty the file contents
+		if err := os.Truncate(filePath, 0); err != nil {
+			slog.Error("Failed to truncate log file", "filePath", filePath, "error", err)
+		}
+		slog.Debug("Truncated log file due to rotation interval", "filePath", filePath)
+	}
+	textLogger.lastWriteTime = now 
+	if _, err := textLogger.handler.WriteString(message); err != nil {
 		slog.Error("Failed to write to log file", "filePath", filePath, "error", err)
 		return
 	}
-
 }
 
 func SetupAllFileLoggers() {
@@ -122,8 +139,8 @@ func SetupAllFileLoggers() {
 }
 
 func CloseAllFileLoggers() {
-	for filePath, file := range fileHandlers {
-		err := file.Close()
+	for filePath, textLogger := range fileHandlers {
+		err := textLogger.handler.Close()
 		if err != nil {
 			slog.Error("Failed to close log file", "filePath", filePath, "error", err)
 			continue