reader: add new reader dirx

cleaner/cleaner.go

@@ -110,7 +110,7 @@ func (c *Cleaner) Name() string {
 	return c.name
 }
 
-func (c *Cleaner) shoudClean(size, count int64) bool {
+func (c *Cleaner) shouldClean(size, count int64) bool {
 	if c.reserveNumber > 0 && count > c.reserveNumber {
 		return true
 	}
@@ -127,14 +127,25 @@ func (c *Cleaner) checkBelong(path string) bool {
 		log.Errorf("GetRealPath for %v error %v", path, err)
 		return false
 	}
-	if c.meta.GetMode() == reader.ModeTailx {
+
+	switch c.meta.GetMode() {
+	case reader.ModeTailx:
 		matched, err := filepath.Match(filepath.Dir(c.logdir), filepath.Dir(path))
 		if err != nil {
-			log.Errorf("checkBelong %v %v err ", c.logdir, path, err)
+			log.Errorf("Failed to check if %q belongs to %q: %v", path, c.logdir, err)
+			return false
+		}
+		return matched
+
+	case reader.ModeDirx:
+		matched, err := filepath.Match(c.logdir, filepath.Dir(path))
+		if err != nil {
+			log.Errorf("Failed to check if %q belongs to %q: %v", path, c.logdir, err)
 			return false
 		}
 		return matched
 	}
+
 	if dir != c.logdir {
 		return false
 	}
@@ -164,7 +175,7 @@ func (c *Cleaner) Clean() (err error) {
 			size += logf.Info.Size()
 			count++
 			// 一旦符合条件，更老的文件必然都要删除
-			if beginClean || c.shoudClean(size, count) {
+			if beginClean || c.shouldClean(size, count) {
 				beginClean = true
 				sig := CleanSignal{
 					Logdir:   filepath.Dir(logf.Path),

mgr/runner.go

@@ -52,10 +52,6 @@ type Runner interface {
 	Status() RunnerStatus
 }
 
-type Resetable interface {
-	Reset() error
-}
-
 type TokenRefreshable interface {
 	TokenRefresh(AuthTokens) error
 }
@@ -544,6 +540,11 @@ func (r *LogExportRunner) readLines(dataSourceTag string) []Data {
 }
 
 func (r *LogExportRunner) Run() {
+	if dr, ok := r.reader.(reader.DaemonReader); ok {
+		if err := dr.Start(); err != nil {
+			log.Errorf("Runner[%v] start reader daemon failed: %v", err)
+		}
+	}
 	if r.cleaner != nil {
 		go r.cleaner.Run()
 	}

reader/autofile/autofile.go

@@ -60,7 +60,7 @@ func matchMode(logpath string) (path, mode string, err error) {
 		return
 	}
 	if fileInfo.IsDir() == true {
-		if shoudUseModeDir(path) {
+		if shouldUseModeDir(path) {
 			mode = reader.ModeDir
 		} else {
 			mode = reader.ModeTailx
@@ -72,7 +72,7 @@ func matchMode(logpath string) (path, mode string, err error) {
 	return
 }
 
-func shoudUseModeDir(logpath string) bool {
+func shouldUseModeDir(logpath string) bool {
 	files, err := ioutil.ReadDir(logpath)
 	if err != nil {
 		log.Warn("read dir %v error %v", logpath, err)

reader/builtin/builtin.go

@@ -5,6 +5,7 @@ import (
 	_ "github.com/qiniu/logkit/reader/autofile"
 	_ "github.com/qiniu/logkit/reader/cloudtrail"
 	_ "github.com/qiniu/logkit/reader/cloudwatch"
+	_ "github.com/qiniu/logkit/reader/dirx"
 	_ "github.com/qiniu/logkit/reader/elastic"
 	_ "github.com/qiniu/logkit/reader/http"
 	_ "github.com/qiniu/logkit/reader/kafka"