chore: add the configuration item maxLineSize (openGemini#592)

Signed-off-by: gueFDF <[email protected]>

Author: gueFDF

config/openGemini.conf

@@ -86,6 +86,7 @@
   # https-private-key = ""
   # time-filter-protection = false
   # parallel-query-in-batch-enabled = true
+  # max-line-size = 65536
 
 [data]
   store-ingest-addr = "{{addr}}:8400"

lib/util/lifted/influx/httpd/config/config.go

@@ -36,7 +36,8 @@ const (
 	// DefaultMaxRowNum is the maximum row number of a query result.
 	DefaultMaxRowNum = 1000000
 
-	DefaultBlockSize = 64 * 1024
+	DefaultBlockSize   = 64 * 1024
+	DefaultMaxLineSize = 1024 * 1024
 )
 
 // Config represents a configuration for a HTTP service.
@@ -87,6 +88,7 @@ type Config struct {
 	ReadBlockSize           toml.Size      `toml:"read-block-size"`
 	TimeFilterProtection    bool           `toml:"time-filter-protection"`
 	CPUThreshold            int            `toml:"cpu-threshold"`
+	MaxLineSize             int            `toml:"max-line-size"`
 }
 
 func CombineDomain(domain, addr string) string {
@@ -135,6 +137,7 @@ func NewConfig() Config {
 		ChunkReaderParallel:     cpu.GetCpuNum(),
 		ReadBlockSize:           toml.Size(DefaultBlockSize),
 		TimeFilterProtection:    false,
+		MaxLineSize:             DefaultMaxLineSize,
 	}
 }
 

lib/util/lifted/influx/httpd/handler.go

@@ -1351,7 +1351,7 @@ func (h *Handler) serveWrite(w http.ResponseWriter, r *http.Request, user meta2.
 		tsMultiplier = 1e9 * 3600
 	}
 
-	ctx := influx.GetStreamContext(body)
+	ctx := influx.GetStreamContext(body, h.Config.MaxLineSize)
 	defer influx.PutStreamContext(ctx)
 
 	var numPtsParse, numPtsInsert int

lib/util/lifted/vm/protoparser/influx/streamparser.go

@@ -91,7 +91,7 @@ func (ctx *streamContext) Read(blockSize int) bool {
 	if ctx.err != nil {
 		return false
 	}
-	ctx.ReqBuf, ctx.tailBuf, ctx.err = ReadLinesBlockExt(ctx.br, ctx.ReqBuf, ctx.tailBuf, maxLineSize, blockSize)
+	ctx.ReqBuf, ctx.tailBuf, ctx.err = ReadLinesBlockExt(ctx.br, ctx.ReqBuf, ctx.tailBuf, ctx.MaxLineSize, blockSize)
 	if ctx.err != nil {
 		if ctx.err != io.EOF {
 			ctx.err = fmt.Errorf("cannot read influx line protocol data: %w", ctx.err)
@@ -111,6 +111,7 @@ type streamContext struct {
 	ErrLock      sync.Mutex
 	UnmarshalErr error // unmarshal points failed, 400 error code
 	CallbackErr  error
+	MaxLineSize  int
 }
 
 func (ctx *streamContext) Error() error {
@@ -129,7 +130,7 @@ func (ctx *streamContext) reset() {
 	ctx.UnmarshalErr = nil
 }
 
-func GetStreamContext(r io.Reader) *streamContext {
+func GetStreamContext(r io.Reader, maxLineSize int) *streamContext {
 	select {
 	case ctx := <-streamContextPoolCh:
 		ctx.br.Reset(r)
@@ -138,11 +139,12 @@ func GetStreamContext(r io.Reader) *streamContext {
 		if v := streamContextPool.Get(); v != nil {
 			ctx := v.(*streamContext)
 			ctx.br.Reset(r)
+			ctx.MaxLineSize = maxLineSize
 			return ctx
 		}
 		return &streamContext{
-			br: bufio.NewReaderSize(r, 64*1024),
-		}
+			br:          bufio.NewReaderSize(r, 64*1024),
+			MaxLineSize: maxLineSize}
 	}
 }
 

lib/util/lifted/vm/protoparser/influx/streamparser_test.go

@@ -0,0 +1,32 @@
+package influx_test
+
+import (
+	"bytes"
+	"testing"
+
+	"github.com/openGemini/openGemini/lib/util/lifted/vm/protoparser/influx"
+)
+
+func TestGetStreamContextMaxLineSize(t *testing.T) {
+	data := "12345678901234567890123456789012" // length == 32
+
+	ctx1 := influx.GetStreamContext(bytes.NewBuffer([]byte(data)), 20)
+	ctx2 := influx.GetStreamContext(bytes.NewBuffer([]byte(data)), 32)
+	ctx3 := influx.GetStreamContext(bytes.NewBuffer([]byte(data)), 33)
+
+	//expect false
+	if ctx1.Read(len(data)) {
+		t.Errorf("expected Read to return false when data length exceeds maxLineSize")
+	}
+
+	//expext true
+	if !ctx2.Read(len(data)) {
+		t.Errorf("expected Read to return true when data length exceeds maxLineSize")
+	}
+
+	//expext true
+	if !ctx3.Read(len(data)) {
+		t.Errorf("expected Read to return true when data length exceeds maxLineSize")
+	}
+
+}