[PDR-12988][fix(ci)]修复ci data race

Author: PapaPiya

mgr/dataflow.go

@@ -37,10 +37,9 @@ const (
 	DefaultRawDataBatchLen = 10
 	RawDataMaxBatchLines   = 100
 	DefaultRawDataSize     = 16 * 1024
+	DefaultRawDataTimeout  = 30
 )
 
-var RawDataTimeOut = 30 * time.Second
-
 // RawData 从 reader 模块中根据 type 获取多条字符串形式的样例日志
 func RawData(readerConfig conf.MapConf) ([]string, error) {
 	defer func() {
@@ -53,6 +52,10 @@ func RawData(readerConfig conf.MapConf) ([]string, error) {
 	}
 
 	runnerName, _ := readerConfig.GetString(GlobalKeyName)
+	rawDataTimeOut, _ := readerConfig.GetIntOr(config.KeyRawDataTimeout, DefaultRawDataTimeout)
+	if rawDataTimeOut < 10 || rawDataTimeOut > 300 {
+		rawDataTimeOut = DefaultRawDataTimeout
+	}
 	configMetaPath := runnerName + "_" + Hash(strconv.FormatInt(time.Now().Unix(), 10))
 	metaPath := filepath.Join(MetaTmp, configMetaPath)
 	log.Debugf("Runner[%v] Using %s as default metaPath", runnerName, metaPath)
@@ -108,15 +111,15 @@ func RawData(readerConfig conf.MapConf) ([]string, error) {
 		}
 		if atomic.LoadInt32(&timeoutStatus) == 1 {
 			if lastErr != nil {
-				readChan <- dataResult{lastErr:lastErr}
+				readChan <- dataResult{lastErr: lastErr}
 				return
 			}
 		}
 
 	}()
 
 	var rawData []string
-	timeout := time.NewTimer(RawDataTimeOut)
+	timeout := time.NewTimer(time.Duration(rawDataTimeOut) * time.Second)
 	defer timeout.Stop()
 	select {
 	case de := <-readChan:

mgr/dataflow_test.go

@@ -135,16 +135,14 @@ func Test_RawDataWithReadData(t *testing.T) {
 	if err != nil {
 		t.Error(err)
 	}
-
-	RawDataTimeOut = 30 * time.Second
+	runnerConf.ReaderConfig[readerconf.KeyRawDataTimeout] = "30"
 	rawData, err := RawData(runnerConf.ReaderConfig)
 	if err != nil {
 		t.Error(err)
 	}
 
 	expected := []string{"{\n  \"logkit\": \"logkit\"\n}"}
 	assert.Equal(t, expected, rawData)
-	RawDataTimeOut = 30 * time.Second
 }
 
 func Test_RawData_DaemonReader(t *testing.T) {
@@ -681,11 +679,11 @@ func Test_RawData_MultiLines(t *testing.T) {
 	assert.Nil(t, err)
 	assert.Equal(t, []string{"abc\n", "abc\n"}, actual)
 
-	RawDataTimeOut = 3 * time.Second
+	readConfig[readerconf.KeyRawDataTimeout] = "3"
 	os.RemoveAll(fileName)
 	createRawDataFile(fileName, "abc\n")
 	actual, err = RawData(readConfig)
-	RawDataTimeOut = 30 * time.Second
+	readConfig[readerconf.KeyRawDataTimeout] = "30"
 	assert.Nil(t, err)
 	assert.Equal(t, []string{"abc\n"}, actual)
 }

mgr/metric_runner_test.go

@@ -200,8 +200,8 @@ func metricRunTest(p *testParam) {
 	base := filepath.Base("")
 	metaPath := "meta/" + runnerName + "_" + Hash(base)
 	t.Log("metaPath: ", metaPath)
-	for i := 0; !utils.IsExist(metaPath) && i < 6; i++ {
-		time.Sleep(500 * time.Millisecond)
+	for i := 0; !utils.IsExist(metaPath) && i < 20; i++ {
+		time.Sleep(1 * time.Second)
 		i++
 	}
 	assert.True(t, utils.IsExist(metaPath))

reader/config/models.go

@@ -10,6 +10,7 @@ const (
 	KeyAuthUsername = "auth_username"
 	KeyAuthPassword = "auth_password"
 
+	KeyRawDataTimeout    = "raw_data_timeout"
 	KeyLogPath           = "log_path"
 	KeyLogPathOld        = "log_path_old"
 	KeyMetaPath          = "meta_path"

reader/tailx/tailx.go

@@ -606,12 +606,19 @@ func (r *Reader) sendError(err error) {
 	if err == nil {
 		return
 	}
+	if r.isStopping() || r.hasStopped() {
+		log.Debugf("Runner[%s] reader %q is not running, send error operation ignored", r.meta.RunnerName, r.Name())
+		return
+	}
 	defer func() {
 		if rec := recover(); rec != nil {
 			log.Errorf("Reader %q was panicked and recovered from %v\nstack: %s", r.Name(), rec, debug.Stack())
 		}
 	}()
-	r.errChan <- err
+	select {
+	case r.errChan <- err:
+	case <- time.After(time.Second):
+	}
 }
 
 // checkExpiredFiles 函数关闭过期的文件，再更新

sender/fault_tolerant/fault_tolerant_test.go

@@ -9,11 +9,10 @@ import (
 	"time"
 	"unicode/utf8"
 
-	"github.com/qiniu/pandora-go-sdk/base/reqerr"
-
 	"github.com/stretchr/testify/assert"
 
 	"github.com/qiniu/log"
+	"github.com/qiniu/pandora-go-sdk/base/reqerr"
 	"github.com/qiniu/pandora-go-sdk/pipeline"
 
 	"github.com/qiniu/logkit/conf"

utils/parse/mutate/keyvalue_test.go

@@ -167,9 +167,10 @@ func Test_parseLine(t *testing.T) {
 		},
 		{
 			line: "fo o=def  = abc ",
-			expectData: []string{
-				"fo o",
-				"def  = abc",
+			expectData: []models.Data{
+				{
+					"fo o": "def  = abc",
+				},
 			},
 			existErr: false,
 			splitter: "=",