Merge pull request #1155 from PapaPiya/fix_dir_dup

修复dir会重复读取的问题

Author: liukai2012

reader/bufreader/bufreader.go

@@ -315,11 +315,7 @@ func (b *BufReader) readSlice(delim []byte) (line []byte, err error) {
 	defer b.mux.Unlock()
 	for {
 		if atomic.LoadInt32(&b.stopped) > 0 {
-			if !IsSelfRunner(b.Meta.RunnerName) {
-				log.Warn("BufReader was stopped while reading...")
-			} else {
-				log.Debug("BufReader was stopped while reading...")
-			}
+			log.Debug("BufReader was stopped while reading...")
 			return
 		}
 		// Search buffer.
@@ -391,7 +387,7 @@ func (b *BufReader) readBytes(delim []byte) ([]byte, error) {
 			}
 			nolog = true
 			lineLen = 0
-			full = make([][]byte, 0, len(full))	// 重新申请空间
+			full = make([][]byte, 0, len(full)) // 重新申请空间
 			continue
 		}
 		buf := make([]byte, fragLen)
@@ -646,7 +642,7 @@ func NewFileDirReader(meta *reader.Meta, conf conf.MapConf) (reader reader.Reade
 	newfileNewLine, _ := conf.GetBoolOr(KeyNewFileNewLine, false)
 	skipFirstLine, _ := conf.GetBoolOr(KeySkipFileFirstLine, false)
 	readSameInode, _ := conf.GetBoolOr(KeyReadSameInode, false)
-	fr, err := seqfile.NewSeqFile(meta, logpath, ignoreHidden, newfileNewLine, ignoreFileSuffix, validFilesRegex, whence, nil, inodeSensitive)
+	fr, err := seqfile.NewSeqFile(meta, logpath, ignoreHidden, newfileNewLine, ignoreFileSuffix, validFilesRegex, whence, inodeSensitive)
 	if err != nil {
 		return
 	}

reader/cloudtrail/cloudtrail.go

@@ -85,7 +85,7 @@ func NewReader(meta *reader.Meta, conf conf.MapConf) (reader.Reader, error) {
 	validFilePattern, _ := conf.GetStringOr(KeyValidFilePattern, "*")
 	bufSize, _ := conf.GetIntOr(KeyBufSize, bufreader.DefaultBufSize)
 	skipFirstLine, _ := conf.GetBoolOr(KeySkipFileFirstLine, false)
-	sf, err := seqfile.NewSeqFile(meta, opts.directory, true, true, ignoredSuffixes, validFilePattern, WhenceOldest, nil, true)
+	sf, err := seqfile.NewSeqFile(meta, opts.directory, true, true, ignoredSuffixes, validFilePattern, WhenceOldest, true)
 	if err != nil {
 		return nil, err
 	}

reader/dirx/dir_reader.go

@@ -351,8 +351,6 @@ type newReaderOptions struct {
 	Whence             string
 	BufferSize         int
 
-	expireMap map[string]int64
-
 	MsgChan chan<- message
 	ErrChan chan<- error
 
@@ -385,7 +383,7 @@ func (drs *dirReaders) NewReader(opts newReaderOptions, notFirstTime bool, maxLi
 			return nil, fmt.Errorf("new extract reader: %v", err)
 		}
 	} else {
-		fr, err := seqfile.NewSeqFile(subMeta, opts.LogPath, opts.IgnoreHidden, opts.NewFileNewLine, opts.IgnoreFileSuffixes, opts.ValidFilesRegex, opts.Whence, opts.expireMap, true)
+		fr, err := seqfile.NewSeqFile(subMeta, opts.LogPath, opts.IgnoreHidden, opts.NewFileNewLine, opts.IgnoreFileSuffixes, opts.ValidFilesRegex, opts.Whence, true)
 		if err != nil {
 			return nil, fmt.Errorf("new sequence file: %v", err)
 		}

reader/dirx/dirx.go

@@ -303,7 +303,6 @@ func (r *Reader) statLogPath() {
 			MsgChan:            r.msgChan,
 			ErrChan:            r.errChan,
 			ReadSameInode:      r.readSameInode,
-			expireMap:          r.expireMap,
 		}, r.notFirstTime, r.maxLineLen)
 		if err != nil {
 			if err == ErrAlreadyExist {

reader/dirx/dirx_test.go

@@ -21,7 +21,7 @@ import (
 )
 
 func createFileWithContent(filepathn, lines string) {
-	file, err := os.OpenFile(filepathn, os.O_CREATE|os.O_WRONLY, DefaultFilePerm)
+	file, err := os.OpenFile(filepathn, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, DefaultFilePerm)
 	if err != nil {
 		log.Error(err)
 		return
@@ -641,7 +641,6 @@ func multiReaderNewestOffsetTest(t *testing.T) {
 
 	createDirWithName(dir1)
 	createFileWithContent(dir1file1, "abc123\nabc124\nabc125\nabc126\nabc127\n")
-	time.Sleep(15 * time.Second)
 	expectResults := map[string]int{
 		"abc\nx\n": 1,
 		"abc\ny\n": 1,
@@ -728,6 +727,10 @@ func multiReaderNewestOffsetTest(t *testing.T) {
 }
 
 func multiReaderSameInodeTest(t *testing.T) {
+	
+}
+
+func TestMultiReaderSameInodeTest(t *testing.T) {
 	dirname := "multiReaderSameInodeTest"
 	dir1 := filepath.Join(dirname, "logs/abc")
 	dir2 := filepath.Join(dirname, "logs/xyz")
@@ -745,8 +748,7 @@ func multiReaderSameInodeTest(t *testing.T) {
 		"abc124\n": 3,
 		"abc125\n": 3,
 		"abc126\n": 3,
-		"abc127\n": 3,
-		"abc128\n": 1,
+		"abc127\n": 2,
 		"abc\nx\n": 1,
 		"abc\ny\n": 1,
 		"abc\nz\n": 1,
@@ -823,7 +825,7 @@ func multiReaderSameInodeTest(t *testing.T) {
 		}
 	}
 	t.Log("Reader has finished reading two")
-	createFileWithContent(dir1file1, "abc123\nabc124\nabc125\nabc126\nabc127\nabc128\n")
+	createFileWithContent(dir1file1, "abc123\nabc124\nabc125\nabc126\n")
 	time.Sleep(5 * time.Second)
 	assert.Equal(t, 2, dr.dirReaders.Num(), "Number of readers")
 

reader/meta.go

@@ -1,6 +1,7 @@
 package reader
 
 import (
+	"bytes"
 	"errors"
 	"fmt"
 	"io/ioutil"
@@ -515,6 +516,25 @@ func (m *Meta) AppendDoneFileInode(path string, inode uint64, offset int64) (err
 	return
 }
 
+func (m *Meta) SyncDoneFileInode(inodeOffset map[string]int64) (err error) {
+	f, err := os.OpenFile(m.DoneFile(), os.O_CREATE|os.O_WRONLY|os.O_TRUNC, DefaultFilePerm)
+	if err != nil {
+		return
+	}
+	defer f.Close()
+
+	var data bytes.Buffer
+	for inodeFile, offset := range inodeOffset {
+		str := strings.Split(inodeFile, "_")
+		if len(str) != 2 {
+			continue
+		}
+		data.WriteString(fmt.Sprintf("%s\t%v\t%v\t%s\n", str[0], str[1], offset, time.Now().Format(time.RFC3339Nano)))
+	}
+	_, err = fmt.Fprintf(f, data.String())
+	return
+}
+
 func (m *Meta) GetDoneFileContent() ([]string, error) {
 	return m.getDoneFileContent()
 }