*: fix several bugs of traffic replay (#698)

Author: djshow832

.golangci.yaml

@@ -30,6 +30,10 @@ issues:
       linters:
         - gosec
       text: "G204:"
+    - path: ".*.go"
+      linters:
+        - gosec
+      text: "G115:"
 
 linters:
   enable:

Makefile

@@ -47,7 +47,7 @@ cmd_%:
 	go build $(BUILDFLAGS) -o $(OUTPUT) $(SOURCE)
 
 golangci-lint:
-	GOBIN=$(GOBIN) go install github.com/golangci/golangci-lint/cmd/golangci-lint@v1.54.1
+	GOBIN=$(GOBIN) go install github.com/golangci/golangci-lint/cmd/golangci-lint@v1.61.0
 
 go-header:
 	GOBIN=$(GOBIN) go install github.com/denis-tingaikin/go-header/cmd/go-header@latest

go.mod

@@ -13,6 +13,7 @@ require (
 	github.com/grpc-ecosystem/go-grpc-middleware v1.3.0
 	github.com/j-keck/arping v1.0.3
 	github.com/klauspost/compress v1.16.6
+	github.com/pelletier/go-toml/v2 v2.0.5
 	github.com/pingcap/kvproto v0.0.0-20231018065736-c0689aded40c
 	github.com/pingcap/sysutil v1.0.0
 	github.com/pingcap/tidb v1.1.0-beta.0.20230103132820-3ccff46aa3bc
@@ -73,7 +74,6 @@ require (
 	github.com/matttproud/golang_protobuf_extensions v1.0.4 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
-	github.com/pelletier/go-toml/v2 v2.0.5 // indirect
 	github.com/pingcap/check v0.0.0-20211026125417-57bd13f7b5f0 // indirect
 	github.com/pingcap/errors v0.11.5-0.20221009092201-b66cddb77c32 // indirect
 	github.com/pingcap/log v1.1.1-0.20221116035753-734d527bc87c // indirect

pkg/manager/vip/network_test.go

@@ -1,7 +1,7 @@
 // Copyright 2024 PingCAP, Inc.
 // SPDX-License-Identifier: Apache-2.0
 
-// go:build linux
+//go:build linux
 
 package vip
 

pkg/proxy/net/capability.go

@@ -96,7 +96,7 @@ func (f Capability) String() string {
 			if cnt > 0 {
 				str.WriteByte('|')
 			}
-			fmt.Fprintf(str, c.Str)
+			fmt.Fprint(str, c.Str)
 			cnt++
 		}
 	}

pkg/proxy/net/mysql.go

@@ -621,13 +621,20 @@ func ParsePrepareStmtResp(resp []byte) (stmtID uint32, paramNum int) {
 	return
 }
 
-func MakeExecuteStmtRequest(stmtID uint32, args []any) ([]byte, error) {
+func MakeExecuteStmtRequest(stmtID uint32, args []any, newParamBound bool) ([]byte, error) {
 	paramNum := len(args)
 	paramTypes := make([]byte, paramNum*2)
 	paramValues := make([][]byte, paramNum)
 	nullBitmap := make([]byte, (paramNum+7)>>3)
-	dataLen := 1 + 4 + 1 + 4 + len(nullBitmap) + 1 + len(paramTypes)
+	dataLen := 1 + 4 + 1 + 4
+	if paramNum > 0 {
+		dataLen += len(nullBitmap) + 1
+	}
 	var newParamBoundFlag byte = 0
+	if newParamBound {
+		newParamBoundFlag = 1
+		dataLen += len(paramTypes)
+	}
 
 	for i := range args {
 		if args[i] == nil {
@@ -636,7 +643,6 @@ func MakeExecuteStmtRequest(stmtID uint32, args []any) ([]byte, error) {
 			continue
 		}
 
-		newParamBoundFlag = 1
 		switch v := args[i].(type) {
 		case int8:
 			paramTypes[i<<1] = fieldTypeTiny
@@ -719,7 +725,7 @@ func MakeExecuteStmtRequest(stmtID uint32, args []any) ([]byte, error) {
 		pos += len(nullBitmap)
 		request[pos] = newParamBoundFlag
 		pos++
-		if newParamBoundFlag == 1 {
+		if newParamBound {
 			copy(request[pos:], paramTypes)
 			pos += len(paramTypes)
 		}
@@ -734,55 +740,62 @@ func MakeExecuteStmtRequest(stmtID uint32, args []any) ([]byte, error) {
 // ParseExecuteStmtRequest parses ComStmtExecute request.
 // NOTICE: the type of returned args may be wrong because it doesn't have the knowledge of real param types.
 // E.g. []byte is returned as string, and int is returned as int32.
-func ParseExecuteStmtRequest(data []byte, paramNum int) (stmtID uint32, args []any, err error) {
-	if len(data) < 1+4+1+4+1 {
-		return 0, nil, errors.WithStack(gomysql.ErrMalformPacket)
+func ParseExecuteStmtRequest(data []byte, paramNum int, paramTypes []byte) (stmtID uint32, args []any, newParamTypes []byte, err error) {
+	if len(data) < 1+4+1+4 {
+		return 0, nil, nil, errors.WithStack(gomysql.ErrMalformPacket)
 	}
 
 	pos := 1
 	stmtID = binary.LittleEndian.Uint32(data[pos : pos+4])
+	// paramNum is contained in the ComStmtPrepare but paramTypes is contained in the first ComStmtExecute (with newParamBoundFlag==1).
+	// If the prepared statement is parsed from the session states, the paramTypes may be empty but the paramNum is not in the session states.
+	// Just return empty args in this case, which is fine currently.
+	if paramNum == 0 {
+		return stmtID, nil, nil, nil
+	}
 	// cursor flag and iteration count
 	pos += 4 + 1 + 4
 	if len(data) < pos+((paramNum+7)>>3)+1 {
-		return 0, nil, errors.WithStack(gomysql.ErrMalformPacket)
+		return 0, nil, nil, errors.WithStack(gomysql.ErrMalformPacket)
 	}
 	nullBitmap := data[pos : pos+((paramNum+7)>>3)]
 	pos += len(nullBitmap)
 	newParamBoundFlag := data[pos]
 	pos += 1
 	args = make([]any, paramNum)
-	if newParamBoundFlag == 0 {
-		return stmtID, args, nil
-	}
 
-	if len(data) < pos+paramNum*2 {
-		return 0, nil, errors.WithStack(gomysql.ErrMalformPacket)
+	if newParamBoundFlag > 0 {
+		if len(data) < pos+paramNum<<1 {
+			return 0, nil, nil, errors.WithStack(gomysql.ErrMalformPacket)
+		}
+		paramTypes = data[pos : pos+paramNum<<1]
+		pos += paramNum << 1
 	}
-	paramTypes := data[pos : pos+paramNum*2]
-	pos += paramNum * 2
 
 	for i := 0; i < paramNum; i++ {
 		if nullBitmap[i/8]&(1<<(uint(i)%8)) > 0 {
 			args[i] = nil
 			continue
 		}
 		switch paramTypes[i<<1] {
+		case fieldTypeNULL:
+			args[i] = nil
 		case fieldTypeTiny:
 			if paramTypes[(i<<1)+1] == 0x80 {
 				args[i] = uint8(data[pos])
 			} else {
 				args[i] = int8(data[pos])
 			}
 			pos += 1
-		case fieldTypeShort:
+		case fieldTypeShort, fieldTypeYear:
 			v := binary.LittleEndian.Uint16(data[pos : pos+2])
 			if paramTypes[(i<<1)+1] == 0x80 {
 				args[i] = v
 			} else {
 				args[i] = int16(v)
 			}
 			pos += 2
-		case fieldTypeLong:
+		case fieldTypeLong, fieldTypeInt24:
 			v := binary.LittleEndian.Uint32(data[pos : pos+4])
 			if paramTypes[(i<<1)+1] == 0x80 {
 				args[i] = v
@@ -804,19 +817,58 @@ func ParseExecuteStmtRequest(data []byte, paramNum int) (stmtID uint32, args []a
 		case fieldTypeDouble:
 			args[i] = math.Float64frombits(binary.LittleEndian.Uint64(data[pos : pos+8]))
 			pos += 8
-		case fieldTypeString:
-			v, _, off, err := ParseLengthEncodedBytes(data[pos:])
+		case fieldTypeDate, fieldTypeTimestamp, fieldTypeDateTime:
+			length := data[pos]
+			pos++
+			switch length {
+			case 0:
+				args[i] = "0000-00-00 00:00:00"
+			case 4:
+				pos, args[i] = BinaryDate(pos, data)
+			case 7:
+				pos, args[i] = BinaryDateTime(pos, data)
+			case 11:
+				pos, args[i] = BinaryTimestamp(pos, data)
+			case 13:
+				pos, args[i] = BinaryTimestampWithTZ(pos, data)
+			default:
+				return 0, nil, nil, errors.WithStack(gomysql.ErrMalformPacket)
+			}
+		case fieldTypeTime:
+			length := data[pos]
+			pos++
+			switch length {
+			case 0:
+				args[i] = "0"
+			case 8:
+				isNegative := data[pos]
+				pos++
+				pos, args[i] = BinaryDuration(pos, data, isNegative)
+			case 12:
+				isNegative := data[pos]
+				pos++
+				pos, args[i] = BinaryDurationWithMS(pos, data, isNegative)
+			default:
+				return 0, nil, nil, errors.WithStack(gomysql.ErrMalformPacket)
+			}
+		case fieldTypeDecimal, fieldTypeNewDecimal, fieldTypeVarChar, fieldTypeString, fieldTypeVarString, fieldTypeBLOB, fieldTypeTinyBLOB,
+			fieldTypeMediumBLOB, fieldTypeLongBLOB, fieldTypeEnum, fieldTypeSet, fieldTypeGeometry, fieldTypeBit, fieldTypeJSON, fieldTypeVector:
+			v, isNull, n, err := ParseLengthEncodedBytes(data[pos:])
 			if err != nil {
-				return 0, nil, errors.Wrapf(err, "parse param %d err", i)
+				return 0, nil, nil, errors.Wrapf(err, "parse param err, type: %d, idx: %d, pos: %d", paramTypes[i<<1], i, pos)
 			}
-			args[i] = hack.String(v)
-			pos += off
+			if isNull {
+				args[i] = nil
+			} else {
+				args[i] = hack.String(v)
+			}
+			pos += n
 		default:
-			return 0, nil, errors.Errorf("unsupported type %d", paramTypes[i<<1])
+			return 0, nil, nil, errors.Errorf("unsupported type %d", paramTypes[i<<1])
 		}
 	}
 
-	return stmtID, args, nil
+	return stmtID, args, paramTypes, nil
 }
 
 func MakeCloseStmtRequest(stmtID uint32) []byte {

pkg/proxy/net/mysql_test.go

@@ -126,19 +126,40 @@ func TestPrepareStmts(t *testing.T) {
 		float64(1.2),
 		nil,
 	}
+	expectedTypes := []byte{
+		fieldTypeNULL, 0,
+		fieldTypeString, 0,
+		fieldTypeTiny, 0x80,
+		fieldTypeShort, 0,
+		fieldTypeLong, 0,
+		fieldTypeLongLong, 0,
+		fieldTypeShort, 0x80,
+		fieldTypeLong, 0x80,
+		fieldTypeLongLong, 0x80,
+		fieldTypeFloat, 0,
+		fieldTypeDouble, 0,
+		fieldTypeNULL, 0,
+	}
 
 	b := MakePrepareStmtRequest("select ?")
 	require.Len(t, b, len("select ?")+1)
 
-	data1, err := MakeExecuteStmtRequest(1, args)
+	data1, err := MakeExecuteStmtRequest(1, args, true)
 	require.NoError(t, err)
 
-	stmtID, pArgs, err := ParseExecuteStmtRequest(data1, len(args))
+	stmtID, pArgs, newParamTypes, err := ParseExecuteStmtRequest(data1, len(args), nil)
 	require.NoError(t, err)
 	require.Equal(t, uint32(1), stmtID)
 	require.EqualValues(t, args, pArgs)
+	require.Equal(t, expectedTypes, newParamTypes)
+
+	data2, err := MakeExecuteStmtRequest(1, pArgs, false)
+	require.NoError(t, err)
+	require.NotEqual(t, data1, data2)
 
-	data2, err := MakeExecuteStmtRequest(1, pArgs)
+	stmtID, pArgs, newParamTypes, err = ParseExecuteStmtRequest(data1, len(args), newParamTypes)
 	require.NoError(t, err)
-	require.Equal(t, data1, data2)
+	require.Equal(t, uint32(1), stmtID)
+	require.EqualValues(t, args, pArgs)
+	require.Equal(t, expectedTypes, newParamTypes)
 }

pkg/proxy/net/protocol.go

@@ -26,6 +26,8 @@ package net
 
 import (
 	"bytes"
+	"encoding/binary"
+	"fmt"
 	"io"
 )
 
@@ -125,12 +127,10 @@ func DumpLengthEncodedInt(buffer []byte, n uint64) []byte {
 	case n <= 0xffffff:
 		return append(buffer, 0xfd, byte(n), byte(n>>8), byte(n>>16))
 
-	case n <= 0xffffffffffffffff:
+	default:
 		return append(buffer, 0xfe, byte(n), byte(n>>8), byte(n>>16), byte(n>>24),
 			byte(n>>32), byte(n>>40), byte(n>>48), byte(n>>56))
 	}
-
-	return buffer
 }
 
 // DumpLengthEncodedString dumps string<int>.
@@ -182,3 +182,67 @@ func Uint64ToBytes(n uint64) []byte {
 		byte(n >> 56),
 	}
 }
+
+func BinaryDate(pos int, paramValues []byte) (int, string) {
+	year := binary.LittleEndian.Uint16(paramValues[pos : pos+2])
+	pos += 2
+	month := paramValues[pos]
+	pos++
+	day := paramValues[pos]
+	pos++
+	return pos, fmt.Sprintf("%04d-%02d-%02d", year, month, day)
+}
+
+func BinaryDateTime(pos int, paramValues []byte) (int, string) {
+	pos, date := BinaryDate(pos, paramValues)
+	hour := paramValues[pos]
+	pos++
+	minute := paramValues[pos]
+	pos++
+	second := paramValues[pos]
+	pos++
+	return pos, fmt.Sprintf("%s %02d:%02d:%02d", date, hour, minute, second)
+}
+
+func BinaryTimestamp(pos int, paramValues []byte) (int, string) {
+	pos, dateTime := BinaryDateTime(pos, paramValues)
+	microSecond := binary.LittleEndian.Uint32(paramValues[pos : pos+4])
+	pos += 4
+	return pos, fmt.Sprintf("%s.%06d", dateTime, microSecond)
+}
+
+func BinaryTimestampWithTZ(pos int, paramValues []byte) (int, string) {
+	pos, timestamp := BinaryTimestamp(pos, paramValues)
+	tzShiftInMin := int16(binary.LittleEndian.Uint16(paramValues[pos : pos+2]))
+	tzShiftHour := tzShiftInMin / 60
+	tzShiftAbsMin := tzShiftInMin % 60
+	if tzShiftAbsMin < 0 {
+		tzShiftAbsMin = -tzShiftAbsMin
+	}
+	pos += 2
+	return pos, fmt.Sprintf("%s%+02d:%02d", timestamp, tzShiftHour, tzShiftAbsMin)
+}
+
+func BinaryDuration(pos int, paramValues []byte, isNegative uint8) (int, string) {
+	sign := ""
+	if isNegative >= 1 {
+		sign = "-"
+	}
+	days := binary.LittleEndian.Uint32(paramValues[pos : pos+4])
+	pos += 4
+	hours := paramValues[pos]
+	pos++
+	minutes := paramValues[pos]
+	pos++
+	seconds := paramValues[pos]
+	pos++
+	return pos, fmt.Sprintf("%s%d %02d:%02d:%02d", sign, days, hours, minutes, seconds)
+}
+
+func BinaryDurationWithMS(pos int, paramValues []byte,
+	isNegative uint8) (int, string) {
+	pos, dur := BinaryDuration(pos, paramValues, isNegative)
+	microSecond := binary.LittleEndian.Uint32(paramValues[pos : pos+4])
+	pos += 4
+	return pos, fmt.Sprintf("%s.%06d", dur, microSecond)
+}

pkg/sqlreplay/capture/capture.go

@@ -277,7 +277,8 @@ func (c *capture) Capture(packet []byte, startTime time.Time, connID uint64, ini
 		// initSession is slow, do not call it in the lock.
 		sql, err := initSession()
 		if err != nil {
-			c.lg.Warn("failed to init session", zap.Error(err))
+			// Maybe the connection is in transaction or closing.
+			c.lg.Debug("failed to init session", zap.Uint64("connID", connID), zap.Error(err))
 			return
 		}
 		initPacket := make([]byte, 0, len(sql)+1)

pkg/sqlreplay/cmd/cmd_test.go

@@ -139,7 +139,7 @@ func TestDigest(t *testing.T) {
 	require.Equal(t, cmd1.Digest(), cmd4.Digest())
 	require.Equal(t, "select ?", cmd4.QueryText())
 
-	data, err := pnet.MakeExecuteStmtRequest(1, []any{1})
+	data, err := pnet.MakeExecuteStmtRequest(1, []any{1}, true)
 	require.NoError(t, err)
 	cmd5 := NewCommand(data, time.Now(), 100)
 	cmd5.PreparedStmt = "select ?"