Merge pull request #46 from hengyoush/fix/redis-parse

[Fix] fix redis message parse issue

Author: hengyoush

agent/conn/conntrack.go

@@ -311,9 +311,16 @@ func (c *Connection4) OnSyscallEvent(data []byte, event *bpf.SyscallEventData, r
 	} else {
 		c.respStreamBuffer.Add(event.SyscallEvent.Ke.Seq, data, event.SyscallEvent.Ke.Ts)
 	}
-
-	c.parseStreamBuffer(c.reqStreamBuffer, protocol.Request, &c.ReqQueue, event.SyscallEvent.Ke.Step)
-	c.parseStreamBuffer(c.respStreamBuffer, protocol.Response, &c.RespQueue, event.SyscallEvent.Ke.Step)
+	reqSteamMessageType := protocol.Request
+	if c.Role == bpf.AgentEndpointRoleTKRoleUnknown {
+		reqSteamMessageType = protocol.Unknown
+	}
+	respSteamMessageType := protocol.Response
+	if c.Role == bpf.AgentEndpointRoleTKRoleUnknown {
+		respSteamMessageType = protocol.Unknown
+	}
+	c.parseStreamBuffer(c.reqStreamBuffer, reqSteamMessageType, &c.ReqQueue, event.SyscallEvent.Ke.Step)
+	c.parseStreamBuffer(c.respStreamBuffer, respSteamMessageType, &c.RespQueue, event.SyscallEvent.Ke.Step)
 	c.StreamEvents.AddSyscallEvent(event)
 
 	parser := c.GetProtocolParser(c.Protocol)

agent/protocol/redis..go

@@ -557,7 +557,7 @@ func (r *RedisStreamParser) ParseStream(streamBuffer *buffer.StreamBuffer, messa
 	seq := streamBuffer.Head().LeftBoundary()
 	ts, ok := streamBuffer.FindTimestampBySeq(seq)
 	decoder := NewBinaryDecoder(head)
-	redisMessage, err := ParseMessage(decoder, ts, seq)
+	parsedMessage, err := ParseMessage(decoder, ts, seq)
 	result := ParseResult{}
 	if err != nil {
 		if errors.Is(err, NotFound) || errors.Is(err, ResourceNotAvailble) {
@@ -572,10 +572,20 @@ func (r *RedisStreamParser) ParseStream(streamBuffer *buffer.StreamBuffer, messa
 			result.ParseState = Success
 		}
 
-		// redisMessage.(*RedisMessage).isReq = messageType == Request
-		redisMessage.(*RedisMessage).seq = seq
-		result.ReadBytes = redisMessage.ByteSize()
-		result.ParsedMessages = []ParsedMessage{redisMessage}
+		redisMessage := parsedMessage.(*RedisMessage)
+		if messageType != Unknown {
+			redisMessage.isReq = messageType == Request
+		} else if redisMessage.isReq {
+			// first byte is `kArrayMarker`, but not sure is a request, guess it only
+			if redisMessage.command == "" {
+				redisMessage.isReq = false
+			} else {
+				redisMessage.isReq = true
+			}
+		}
+		redisMessage.seq = seq
+		result.ReadBytes = parsedMessage.ByteSize()
+		result.ParsedMessages = []ParsedMessage{parsedMessage}
 	}
 
 	return result