Merge pull request #933 from wonderflow/fix1

support ignoreHidden for extract reader  && fix dead lock on dirx close

Author: wonderflow

reader/autofile/autofile.go

@@ -14,6 +14,7 @@ import (
 	"github.com/qiniu/logkit/reader"
 	"github.com/qiniu/logkit/reader/bufreader"
 	"github.com/qiniu/logkit/reader/config"
+	"github.com/qiniu/logkit/reader/dirx"
 	"github.com/qiniu/logkit/reader/tailx"
 )
 
@@ -39,6 +40,9 @@ func NewReader(meta *reader.Meta, conf conf.MapConf) (r reader.Reader, err error
 		return bufreader.NewFileDirReader(meta, conf)
 	case config.ModeFile:
 		return bufreader.NewSingleFileReader(meta, conf)
+	case config.ModeDirx:
+		conf[config.KeyLogPath] = logpath
+		return dirx.NewReader(meta, conf)
 	default:
 		err = errors.New("can not find property mode for this path " + logpath)
 	}
@@ -51,6 +55,15 @@ func matchMode(logpath string) (path, mode string, err error) {
 		logpath = filepath.Dir(logpath)
 	}
 	path = logpath
+	if strings.HasSuffix(logpath, ".tar.gz") || strings.HasSuffix(logpath, ".tar") || strings.HasSuffix(logpath, ".zip") {
+		mode = config.ModeDirx
+		return
+	}
+	if strings.HasSuffix(logpath, ".gz") {
+		mode = config.ModeTailx
+		return
+	}
+
 	//path with * matching tailx mode
 	matchTailx := strings.Contains(logpath, "*")
 	if matchTailx == true {

reader/bufreader/bufreader.go

@@ -55,7 +55,7 @@ type LastSync struct {
 type BufReader struct {
 	stopped       int32
 	buf           []byte
-	mutiLineCache []string
+	mutiLineCache *LineCache
 	rd            reader.FileReader // reader provided by the client
 	r, w          int               // buf read and write positions
 	err           error
@@ -136,6 +136,7 @@ func NewReaderSize(rd reader.FileReader, meta *reader.Meta, size int) (*BufReade
 	r := new(BufReader)
 	r.stopped = 0
 	r.reset(make([]byte, size), rd)
+	r.mutiLineCache = NewLineCache()
 
 	r.Meta = meta
 	encodingWay := r.Meta.GetEncodingWay()
@@ -163,7 +164,7 @@ func NewReaderSize(rd reader.FileReader, meta *reader.Meta, size int) (*BufReade
 		}
 	}
 	if len(linesbytes) > 0 {
-		r.mutiLineCache = append(r.mutiLineCache, string(linesbytes))
+		r.mutiLineCache.Append(string(linesbytes))
 	}
 	return r, nil
 }
@@ -183,7 +184,7 @@ func (b *BufReader) reset(buf []byte, r reader.FileReader) {
 		rd:            r,
 		lastByte:      -1,
 		lastRuneSize:  -1,
-		mutiLineCache: make([]string, 0, 16),
+		mutiLineCache: NewLineCache(),
 		lastSync:      LastSync{},
 		mux:           sync.Mutex{},
 		statsLock:     sync.RWMutex{},
@@ -242,18 +243,24 @@ func (b *BufReader) fill() {
 			panic(errNegativeRead)
 		}
 		if b.latestSource != b.rd.Source() {
-			//这个情况表示文件的数据源出现了变化，在buf中已经出现了2个数据源的数据，要定位是哪个位置的数据出现的分隔
-			if rc, ok := b.rd.(seqfile.NewLineBytesRecorder); ok {
-				SIdx := rc.NewLineBytesIndex()
+			//这个情况表示文件的数据源出现了变化，在buf中已经出现了至少2个数据源的数据，要定位是哪个位置的数据出现的分隔
+			if rc, ok := b.rd.(reader.NewSourceRecorder); ok {
+				SIdx := rc.NewSourceIndex()
 				for _, v := range SIdx {
-					// 从 NewLineBytesIndex 函数中返回的index值就是本次读取的批次中上一个DataSource的数据量，加上b.w就是上个DataSource的整体数据
+					// 从 NewSourceIndex 函数中返回的index值就是本次读取的批次中上一个DataSource的数据量，加上b.w就是上个DataSource的整体数据
 					b.lastRdSource = append(b.lastRdSource, reader.SourceIndex{
 						Source: v.Source,
 						Index:  b.w + v.Index,
 					})
 				}
-				b.latestSource = b.rd.Source()
+			} else {
+				//如果没实现这个接口，那么就认为到上次读到的为止都是前一次source的文件
+				b.lastRdSource = append(b.lastRdSource, reader.SourceIndex{
+					Source: b.latestSource,
+					Index:  b.w,
+				})
 			}
+			b.latestSource = b.rd.Source()
 		}
 
 		b.w += n
@@ -404,24 +411,24 @@ func (b *BufReader) ReadPattern() (string, error) {
 		line, err := b.ReadString('\n')
 		//读取到line的情况
 		if len(line) > 0 {
-			if len(b.mutiLineCache) <= 0 {
-				b.mutiLineCache = []string{line}
+			if b.mutiLineCache.Size() <= 0 {
+				b.mutiLineCache.Set([]string{line})
 				continue
 			}
 			//匹配行首，成功则返回之前的cache，否则加入到cache，返回空串
 			if b.multiLineRegexp.Match([]byte(line)) {
 				tmp := line
-				line = string(b.FormMutiLine())
-				b.mutiLineCache = make([]string, 0, 16)
-				b.mutiLineCache = append(b.mutiLineCache, tmp)
+				line = string(b.mutiLineCache.Combine())
+				b.mutiLineCache.Set(make([]string, 0, 16))
+				b.mutiLineCache.Append(tmp)
 				return line, err
 			}
-			b.mutiLineCache = append(b.mutiLineCache, line)
+			b.mutiLineCache.Append(line)
 			maxTimes = 0
 		} else { //读取不到日志
 			if err != nil {
-				line = string(b.FormMutiLine())
-				b.mutiLineCache = make([]string, 0, 16)
+				line = string(b.mutiLineCache.Combine())
+				b.mutiLineCache.Set(make([]string, 0, 16))
 				return line, err
 			}
 			maxTimes++
@@ -432,36 +439,16 @@ func (b *BufReader) ReadPattern() (string, error) {
 			}
 		}
 		//对于读取到了Cache的情况，继续循环，直到超过最大限制
-		if b.calcMutiLineCache() > MaxHeadPatternBufferSize {
-			line = string(b.FormMutiLine())
-			b.mutiLineCache = make([]string, 0, 16)
+		if b.mutiLineCache.TotalLen() > MaxHeadPatternBufferSize {
+			line = string(b.mutiLineCache.Combine())
+			b.mutiLineCache.Set(make([]string, 0, 16))
 			return line, err
 		}
 	}
 }
 
 func (b *BufReader) FormMutiLine() []byte {
-	if len(b.mutiLineCache) <= 0 {
-		return make([]byte, 0)
-	}
-	n := 0
-	for i := 0; i < len(b.mutiLineCache); i++ {
-		n += len(b.mutiLineCache[i])
-	}
-
-	xb := make([]byte, n)
-	bp := copy(xb, b.mutiLineCache[0])
-	for _, s := range b.mutiLineCache[1:] {
-		bp += copy(xb[bp:], s)
-	}
-	return xb
-}
-
-func (b *BufReader) calcMutiLineCache() (ret int) {
-	for _, v := range b.mutiLineCache {
-		ret += len(v)
-	}
-	return
+	return b.mutiLineCache.Combine()
 }
 
 //ReadLine returns a string line as a normal Reader
@@ -561,7 +548,7 @@ func (b *BufReader) ReadDone() bool {
 func (b *BufReader) SyncMeta() {
 	b.mux.Lock()
 	defer b.mux.Unlock()
-	linecache := string(b.FormMutiLine())
+	linecache := string(b.mutiLineCache.Combine())
 	//把linecache也缓存
 	if b.lastSync.cache != linecache || b.lastSync.buf != string(b.buf) || b.r != b.lastSync.r || b.w != b.lastSync.w {
 		log.Debugf("Runner[%v] %v sync meta started, linecache [%v] buf [%v] （%v %v）", b.Meta.RunnerName, b.Name(), linecache, string(b.buf), b.r, b.w)

reader/bufreader/linecache.go

@@ -0,0 +1,60 @@
+package bufreader
+
+import (
+	"sync"
+)
+
+type LineCache struct {
+	lines []string
+	lock  *sync.RWMutex
+}
+
+func NewLineCache() *LineCache {
+	return &LineCache{
+		lines: make([]string, 0, 16),
+		lock:  new(sync.RWMutex),
+	}
+}
+
+func (l *LineCache) Size() int {
+	l.lock.RLock()
+	defer l.lock.RUnlock()
+	return len(l.lines)
+}
+
+func (l *LineCache) Set(r []string) {
+	l.lock.Lock()
+	defer l.lock.Unlock()
+	l.lines = r
+}
+
+func (l *LineCache) Append(r string) {
+	l.lock.Lock()
+	defer l.lock.Unlock()
+	l.lines = append(l.lines, r)
+}
+
+func (l *LineCache) TotalLen() int {
+	l.lock.RLock()
+	defer l.lock.RUnlock()
+	var ret int
+	for _, v := range l.lines {
+		ret += len(v)
+	}
+	return ret
+}
+
+func (l *LineCache) Combine() []byte {
+	if l.Size() <= 0 {
+		return make([]byte, 0)
+	}
+	n := l.TotalLen()
+	xb := make([]byte, n)
+	l.lock.RLock()
+	defer l.lock.RUnlock()
+	bp := copy(xb, l.lines[0])
+	for _, s := range l.lines[1:] {
+		bp += copy(xb[bp:], s)
+	}
+	return xb
+}

reader/bufreader/linecache_test.go

@@ -0,0 +1,23 @@
+package bufreader
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestLinecahe(t *testing.T) {
+	l := NewLineCache()
+	l.Append("abc")
+	assert.Equal(t, 3, l.TotalLen())
+	l.Append("456")
+	assert.Equal(t, 6, l.TotalLen())
+	assert.Equal(t, 2, l.Size())
+	assert.Equal(t, "abc456", string(l.Combine()))
+	l.Set([]string{"bbb"})
+	assert.Equal(t, 3, l.TotalLen())
+	l.Append("haha")
+	assert.Equal(t, 7, l.TotalLen())
+	assert.Equal(t, 2, l.Size())
+	assert.Equal(t, "bbbhaha", string(l.Combine()))
+}

reader/dirx/dir_reader.go

@@ -24,6 +24,8 @@ import (
 	. "github.com/qiniu/logkit/utils/models"
 )
 
+var ErrAlreadyExist = errors.New("runner already exist")
+
 type dirReader struct {
 	status        int32 // Note: 原子操作
 	inactive      int32 // Note: 原子操作，当 inactive>0 时才会被 expire 回收
@@ -180,11 +182,17 @@ func HasDirExpired(dir string, expire time.Duration) bool {
 	return latestModTime.Add(expire).Before(time.Now())
 }
 
+//对于读完的直接认为过期，因为不会追加新数据
+func (dr *dirReader) ReadDone() bool {
+	return dr.br.ReadDone()
+}
+
 func (dr *dirReader) HasExpired(expire time.Duration) bool {
-	if dr.br.ReadDone() {
-		//对于读完的直接认为过期，因为不会追加新数据
-		return true
+	// 如果过期时间为 0，则永不过期
+	if expire.Nanoseconds() == 0 {
+		return false
 	}
+
 	// 在非 inactive 的情况下，数据尚未读完，有必要先继续处理
 	return atomic.LoadInt32(&dr.inactive) > 0 && HasDirExpired(dr.logPath, expire)
 }
@@ -206,6 +214,7 @@ func (dr *dirReader) SyncMeta() string {
 
 func (dr *dirReader) Close() error {
 	defer log.Warnf("Runner[%v] log path[%v] reader has closed", dr.runnerName, dr.originalPath)
+	dr.SyncMeta()
 	err := dr.br.Close()
 	if atomic.CompareAndSwapInt32(&dr.status, StatusRunning, StatusStopping) {
 		log.Warnf("Runner[%v] log path[%v] reader is closing", dr.runnerName, dr.originalPath)
@@ -304,7 +313,7 @@ func (drs *dirReaders) NewReader(opts newReaderOptions, notFirstTime bool) (*dir
 	}
 	var fri reader.FileReader
 	if reader.CompressedFile(opts.LogPath) {
-		fri, err = extract.NewReader(subMeta, opts.LogPath)
+		fri, err = extract.NewReader(subMeta, opts.LogPath, extract.Opts{IgnoreHidden: opts.IgnoreHidden, NewFileNewLine: opts.NewFileNewLine, IgnoreFileSuffixes: opts.IgnoreFileSuffixes, ValidFilesRegex: opts.ValidFilesRegex})
 		if err != nil {
 			return nil, fmt.Errorf("new extract reader: %v", err)
 		}
@@ -338,6 +347,11 @@ func (drs *dirReaders) NewReader(opts newReaderOptions, notFirstTime bool) (*dir
 	drs.lock.Lock()
 	defer drs.lock.Unlock()
 
+	//double check
+	if _, ok := drs.readers[opts.LogPath]; ok {
+		return nil, ErrAlreadyExist
+	}
+
 	dr.readcache = drs.cachedLines[opts.LogPath]
 	opts.Meta.AddSubMeta(opts.LogPath, subMeta)
 	drs.readers[opts.LogPath] = dr
@@ -358,7 +372,7 @@ func (drs *dirReaders) checkExpiredDirs() {
 
 	var expiredDirs []string
 	for logPath, dr := range drs.readers {
-		if dr.HasExpired(drs.expire) {
+		if dr.HasExpired(drs.expire) || (drs.expireDelete && dr.ReadDone()) {
 			if err := dr.Close(); err != nil {
 				log.Errorf("Failed to close log path[%v] reader: %v", logPath, err)
 			}
@@ -380,10 +394,9 @@ func (drs *dirReaders) checkExpiredDirs() {
 }
 
 func (drs *dirReaders) getReaders() []*dirReader {
+	readers := make([]*dirReader, 0, drs.Num())
 	drs.lock.RLock()
 	defer drs.lock.RUnlock()
-
-	readers := make([]*dirReader, 0, drs.Num())
 	for _, dr := range drs.readers {
 		readers = append(readers, dr)
 	}

reader/dirx/dirx.go

@@ -302,6 +302,9 @@ func (r *Reader) statLogPath() {
 			expireMap:          r.expireMap,
 		}, r.notFirstTime)
 		if err != nil {
+			if err == ErrAlreadyExist {
+				continue
+			}
 			err = fmt.Errorf("create new reader for log path %q failed: %v", logPath, err)
 			r.sendError(err)
 			log.Errorf("Runner[%v] %v, ignored this path", r.meta.RunnerName, err)