Merge pull request #1161 from asiocity/PDR-13603

[PDR-13603][bug] 修复 s3 接口无法对接其他平台问题

Author: liukai2012

mgr/dataflow_test.go

@@ -679,7 +679,7 @@ func Test_RawData_MultiLines(t *testing.T) {
 	assert.Nil(t, err)
 	assert.Equal(t, []string{"abc\n", "abc\n"}, actual)
 
-	readConfig[readerconf.KeyRawDataTimeout] = "3"
+	readConfig[readerconf.KeyRawDataTimeout] = "40"
 	os.RemoveAll(fileName)
 	createRawDataFile(fileName, "abc\n")
 	actual, err = RawData(readConfig)

mgr/runner_test.go

@@ -1469,7 +1469,7 @@ func TestAddDatatags(t *testing.T) {
 	assert.NoError(t, err)
 	go rr.Run()
 
-	time.Sleep(2 * time.Second)
+	time.Sleep(60 * time.Second)
 	data, err := ioutil.ReadFile("./TestAddDatatags/filesend.json")
 	assert.Nil(t, err)
 	var res []Data
@@ -1539,7 +1539,7 @@ func TestRunWithExtra(t *testing.T) {
 	assert.NoError(t, err)
 	go rr.Run()
 
-	time.Sleep(2 * time.Second)
+	time.Sleep(60 * time.Second)
 	data, err := ioutil.ReadFile("./TestRunWithExtra/filesend.json")
 	assert.Nil(t, err)
 	var res []Data
@@ -1671,7 +1671,7 @@ func TestRunWithDataSourceFail(t *testing.T) {
 	assert.NotNil(t, rr)
 	go rr.Run()
 
-	time.Sleep(2 * time.Second)
+	time.Sleep(60 * time.Second)
 	data, err := ioutil.ReadFile("./TestRunWithDataSourceFail/filesend.json")
 	assert.Nil(t, err)
 	var res []Data
@@ -2076,21 +2076,21 @@ func TestTailxCleaner(t *testing.T) {
 	assert.NotNil(t, rr)
 	go rr.Run()
 
-	time.Sleep(2 * time.Second)
+	time.Sleep(60 * time.Second)
 
 	logPatha1 := filepath.Join(dira, "a.log.1")
 	assert.NoError(t, os.Rename(logPatha, logPatha1))
 
 	assert.NoError(t, ioutil.WriteFile(logPatha, []byte("bbbb\n"), 0666))
 
-	time.Sleep(5 * time.Second)
+	time.Sleep(60 * time.Second)
 
 	logPatha2 := filepath.Join(dira, "a.log.2")
 	assert.NoError(t, os.Rename(logPatha, logPatha2))
 
 	assert.NoError(t, ioutil.WriteFile(logPatha, []byte("cccc\n"), 0666))
 
-	time.Sleep(2 * time.Second)
+	time.Sleep(60 * time.Second)
 
 	assert.NotNil(t, rr.Cleaner())
 
@@ -2113,6 +2113,7 @@ DONE:
 			break
 		}
 	}
+	time.Sleep(50 * time.Second)
 	assert.Equal(t, 1, ret)
 }
 

reader/bufreader/bufreader.go

@@ -54,16 +54,17 @@ type LastSync struct {
 
 // BufReader implements buffering for an FileReader object.
 type BufReader struct {
-	stopped       int32
-	buf           []byte
-	delim         []byte
-	mutiLineCache *LineCache
-	rd            reader.FileReader // reader provided by the client
-	r, w          int               // buf read and write positions
-	err           error
-	lastByte      int
-	lastRuneSize  int
-	lastSync      LastSync
+	stopped          int32
+	buf              []byte
+	delim            []byte
+	mutiLineCache    *LineCache
+	waitForWholeLine bool              //readWholeLine
+	rd               reader.FileReader // reader provided by the client
+	r, w             int               // buf read and write positions
+	err              error
+	lastByte         int
+	lastRuneSize     int
+	lastSync         LastSync
 
 	runTime reader.RunTime
 
@@ -88,7 +89,7 @@ type BufReader struct {
 const minReadBufferSize = 16
 
 //最大连续读到空的尝试次数
-const maxConsecutiveEmptyReads = 10
+const maxConsecutiveEmptyReads = 40
 
 // NewReaderSize returns a new Reader whose buffer has at least the specified
 // size. If the argument FileReader is already a Reader with large enough
@@ -189,6 +190,11 @@ func (b *BufReader) SetMode(mode string, v interface{}) (err error) {
 	return
 }
 
+func (b *BufReader) SetWaitFlagForWholeLine() {
+	b.waitForWholeLine = true
+	return
+}
+
 func (b *BufReader) SetRunTime(mode string, v interface{}) (err error) {
 	b.runTime, err = reader.ParseRunTimeWithMode(mode, v)
 	return err
@@ -280,13 +286,25 @@ func (b *BufReader) fill() {
 		}
 
 		b.w += n
+
+		if err == io.EOF && b.waitForWholeLine {
+			if i == 1 { //when last attempts,return err info;
+				b.err = err
+				return
+			}
+
+			time.Sleep(1 * time.Second)
+			continue
+		}
+
 		if err != nil {
 			b.err = err
 			return
 		}
 		if n > 0 {
 			return
 		}
+
 	}
 	b.err = io.ErrNoProgress
 }
@@ -666,7 +684,9 @@ func NewSingleFileReader(meta *reader.Meta, conf conf.MapConf) (reader reader.Re
 		return
 	}
 	maxLineLen, _ := conf.GetInt64Or(KeyRunnerMaxLineLen, 0)
-	return NewReaderSize(fr, meta, bufSize, maxLineLen)
+	r, err := NewReaderSize(fr, meta, bufSize, maxLineLen)
+	r.SetWaitFlagForWholeLine()
+	return r, err
 }
 
 func init() {

reader/cloudtrail/cloudtrail.go

@@ -248,12 +248,12 @@ func newSyncManager(meta *reader.Meta, opts *syncOptions) (*syncManager, error)
 
 func makeSyncSource(bucket, prefix string) string {
 	if prefix == "" {
-		return fmt.Sprintf("s3://%s", bucket)
+		return fmt.Sprintf("%s", bucket)
 	}
 	if strings.HasPrefix(prefix, "/") {
-		return fmt.Sprintf("s3://%s%s", bucket, prefix)
+		return fmt.Sprintf("%s%s", bucket, prefix)
 	}
-	return fmt.Sprintf("s3://%s/%s", bucket, prefix)
+	return fmt.Sprintf("%s/%s", bucket, prefix)
 }
 
 func (mgr *syncManager) startSync() {
@@ -319,9 +319,6 @@ func newSyncRunner(ctx *syncContext, quitChan chan struct{}) *syncRunner {
 }
 
 func (s *syncRunner) Sync() error {
-	if !validSource(s.source) {
-		return fmt.Errorf("invalid sync source %q", s.source)
-	}
 	if !validTarget(s.target) {
 		return fmt.Errorf("invalid sync target %q", s.target)
 	}
@@ -457,10 +454,20 @@ func (r *s3Url) keys() []string {
 
 func lookupBucket(bucketName string, auth aws.Auth, region string, endpoint string) (*s3.Bucket, error) {
 	log.Infof("looking for bucket %q in region %q", bucketName, region)
-	rg := aws.Regions[region]
-	if endpoint != "" {
-		rg.S3Endpoint = endpoint
+
+	rg, ok := aws.Regions[region]
+	if !ok {
+		rg = aws.Region{
+			Name:       region,
+			S3Endpoint: endpoint,
+		}
+	} else {
+		// replace default endpoint
+		if endpoint != "" {
+			rg.S3Endpoint = endpoint
+		}
 	}
+
 	s3 := s3.New(auth, rg)
 	bucket := s3.Bucket(bucketName)
 	_, err := bucket.List("", "", "", 0)

reader/dirx/dirx_test.go

@@ -1079,6 +1079,7 @@ func readerExpireDeleteTarTest(t *testing.T) {
 		}
 	}
 	t.Log("maxNum ", maxNum, "emptyNum", emptyNum)
+	time.Sleep(60 * time.Second)
 	assert.EqualValues(t, len(expectResults), len(actualResults))
 	for k, v := range expectResults {
 		actualV, ok := actualResults[k]