Increase the maximum read size of a single data packet

vearne · Aug 7, 2024 · ffa4f13 · ffa4f13
1 parent 5977fb3
commit ffa4f13
Show file tree

Hide file tree

Showing 3 changed files with 14 additions and 4 deletions.
diff --git a/http2/http2.go b/http2/http2.go
@@ -9,6 +9,7 @@ import (
 	"golang.org/x/net/http2/hpack"
 	"google.golang.org/protobuf/encoding/protojson"
 	"google.golang.org/protobuf/proto"
+	"strings"
 	"time"
 
 	"github.com/google/uuid"
@@ -165,7 +166,7 @@ func (s *Stream) toMsg(finder *PBMessageFinder) *protocol.Message {
 	if codecType == CodecProtobuf {
 		if len(s.Method) <= 0 {
 			slog.Error("method is empty, this is illegal")
-		} else {
+		} else if !strings.Contains(s.Method, "grpc.reflection") {
 			// Note: Temporarily only handle the case where the encoding method is Protobuf
 			pbMsg := finder.FindMethodInputWithCache(s.Method)
 			err := proto.Unmarshal(s.DataBuf.Bytes(), pbMsg)

diff --git a/http2/processor.go b/http2/processor.go
@@ -51,6 +51,9 @@ func (p *Processor) ProcessTCPPkg() {
 			continue
 		}
 
+		dc := pkg.DirectConn()
+		slog.Debug("Connection:%v, seq:%v, length:%v", &dc, pkg.TCP.Seq, len(payload))
+
 		for len(payload) >= HeaderSize {
 			f, err = ParseFrameBase(payload)
 			if err != nil {
@@ -60,8 +63,8 @@ func (p *Processor) ProcessTCPPkg() {
 
 			dc := pkg.DirectConn()
 			f.DirectConn = &dc
-			slog.Debug("Connection:%v, seq:%v, FrameType:%v, length:%v, streamID:%v",
-				f.DirectConn, pkg.TCP.Seq, GetFrameType(f.Type), f.Length, f.StreamID)
+			slog.Debug("Connection:%v, seq:%v, FrameType:%v, length:%v, len(payload):%v, streamID:%v",
+				f.DirectConn, pkg.TCP.Seq, GetFrameType(f.Type), f.Length, len(f.Payload), f.StreamID)
 
 			var ok bool
 			if _, ok = p.ConnRepository[dc]; !ok {
@@ -125,6 +128,9 @@ func (p *Processor) processFrameData(f *FrameBase) {
 		return
 	}
 
+	slog.Debug("processFrameData, Padded:%v, PadLength:%v, EndStream:%v, len(fd.Data):%v",
+		fd.Padded, fd.PadLength, fd.EndStream, len(fd.Data))
+
 	hc := p.ConnRepository[*f.DirectConn]
 	// Set the state of the stream
 	index := f.StreamID % StreamArraySize
@@ -136,6 +142,7 @@ func (p *Processor) processFrameData(f *FrameBase) {
 		msg, _ := fd.ParseGRPCMessage()
 		// Compression is turned on
 		if msg.PayloadFormat == compressionMade {
+			slog.Debug("msg.PayloadFormat == compressionMade")
 			// only support gzip
 			gzipReader, err = gzip.NewReader(bytes.NewReader(msg.EncodedMessage))
 			if err != nil {
@@ -149,6 +156,7 @@ func (p *Processor) processFrameData(f *FrameBase) {
 			}
 		}
 
+		slog.Debug("len(msg.EncodedMessage):%v", len(msg.EncodedMessage))
 		_, err = stream.DataBuf.Write(msg.EncodedMessage)
 		if err != nil {
 			slog.Error("processFrameData, gunzip error:%v", err)
@@ -305,6 +313,7 @@ func (f *PBMessageFinder) FindMethodInputWithCache(svcAndMethod string) proto.Me
 	m, ok := f.symbolMsg[svcAndMethod]
 	f.cacheMu.RUnlock()
 	if ok {
+		slog.Debug("FindMethodInputWithCache,svcAndMethod:%v, hit cache", svcAndMethod)
 		return m
 	}
 

diff --git a/plugin/input_raw.go b/plugin/input_raw.go
@@ -17,7 +17,7 @@ import (
 )
 
 const (
-	snapshotLen int32         = 1024
+	snapshotLen int32         = 20000
 	promiscuous bool          = false
 	timeout     time.Duration = 5 * time.Second
 )