Merge pull request #602 from redHJ/rawdata

1. close reader 2. raw data with mock meta

Author: wonderflow

mgr/dataflow.go

@@ -4,10 +4,15 @@ import (
 "errors"
 "fmt"
 "io"
+"os"
+"path"
+"strconv"
 "strings"
 "time"
 
 "github.com/json-iterator/go"
+
+"github.com/qiniu/log"
 "github.com/qiniu/pandora-go-sdk/base/reqerr"
 
 "github.com/qiniu/logkit/conf"
@@ -20,19 +25,31 @@ import (
 . "github.com/qiniu/logkit/utils/models"
 )
 
-const DefaultTryTimes = 3
+const (
+DefaultTryTimes = 3
+MetaTmp = "meta_tmp/"
+)
 
 // RawData 从 reader 模块中根据 type 获取字符串形式的样例日志
 func RawData(readerConfig conf.MapConf) (string, error) {
 if readerConfig == nil {
 return "", fmt.Errorf("reader config cannot be empty")
 }
 
+runnerName, _ := readerConfig.GetString(GlobalKeyName)
+configMetaPath := runnerName + "_" + Hash(strconv.FormatInt(time.Now().Unix(), 10))
+metaPath := path.Join(MetaTmp, configMetaPath)
+log.Debugf("Runner[%v] Using %s as default metaPath", runnerName, metaPath)
+readerConfig[reader.KeyMetaPath] = metaPath
+
 rd, err := reader.NewReader(readerConfig, true)
 if err != nil {
 return "", err
 }
-defer rd.Close()
+defer func() {
+rd.Close()
+os.RemoveAll(metaPath)
+}()
 
 var rawData string
 

mgr/runner.go

@@ -232,16 +232,16 @@ func NewLogExportRunner(rc RunnerConfig, cleanChan chan<- cleaner.CleanSignal, r
 if err != nil {
 return nil, err
 }
+defer func() {
+if err != nil && rd != nil {
+rd.Close()
+}
+}()
 if len(rc.CleanerConfig) > 0 {
 rd, err = rr.NewReaderWithMeta(rc.ReaderConfig, meta, false)
 if err != nil {
 return nil, err
 }
-defer func() {
-if err != nil {
-rd.Close()
-}
-}()
 cl, err = cleaner.NewCleaner(rc.CleanerConfig, meta, cleanChan, meta.LogPath())
 if err != nil {
 return nil, err

reader/reader.go

@@ -265,12 +265,15 @@ func (reg *Registry) NewReaderWithMeta(conf conf.MapConf, meta *Meta, errDirectR
 
 reader, err = constructor(meta, conf)
 if err != nil {
-return
+return nil, err
 }
 if headPattern != "" {
 err = reader.SetMode(ReadModeHeadPatternString, headPattern)
+if err != nil {
+return nil, err
+}
 }
-return
+return reader, nil
 }
 
 func NewFileDirReader(meta *Meta, conf conf.MapConf) (reader Reader, err error) {