revert globalreader

Author: ayushaga14

ebpf/bpfwrapper/perfbufferreaders.go

@@ -36,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, 100)
+	probeChannel.eventChannel = make(chan []byte)
 	probeChannel.lostEventsChannel = make(chan uint64)
 
 	table := bcc.NewTable(module.TableId(probeChannel.name), module)

trafficUtil/kafkaUtil/parser.go

@@ -40,10 +40,7 @@ var (
 		"TRACE":   true,
 		"TRACK":   true,
 		"PATCH":   true}
-	DebugStrings     = []string{}
-	globalReader     = &bytes.Reader{}
-	globalReaderLock sync.Mutex
-
+	DebugStrings = []string{}
 )
 
 const ONE_MINUTE = 60
@@ -133,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)
@@ -185,12 +182,7 @@ func ParseAndProduce(receiveBuffer []byte, sentBuffer []byte, sourceIp string, d
 		slog.Debug("ParseAndProduce", "receiveBuffer", string(receiveBuffer), "sentBuffer", string(sentBuffer))
 	}
 
-	reader := func() *bufio.Reader {
-		globalReaderLock.Lock()
-		defer globalReaderLock.Unlock()
-		globalReader.Reset(receiveBuffer)
-		return bufio.NewReader(globalReader)
-	}()
+	reader := bufio.NewReader(bytes.NewReader(receiveBuffer))
 	i := 0
 	requests := []http.Request{}
 	requestsContent := []string{}
@@ -222,12 +214,7 @@ func ParseAndProduce(receiveBuffer []byte, sentBuffer []byte, sourceIp string, d
 		return
 	}
 
-	reader = func() *bufio.Reader {
-		globalReaderLock.Lock()
-		defer globalReaderLock.Unlock()
-		globalReader.Reset(sentBuffer)
-		return bufio.NewReader(globalReader)
-	}()
+	reader = bufio.NewReader(bytes.NewReader(sentBuffer))
 	i = 0
 
 	responses := []http.Response{}
@@ -474,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)