flume-ng-taildirectory-source 修改调试可用

浏览数：70 / 时间：2015年06月09日

由于flume-ng至1.5版本仍旧没有稳定可用的类似flume-og中的taildir的功能，所以从git中https://github.com/jinoos/flume-ng-extends找了一个别人针对flume-ng实现的的taildir这个按照github上他自己说明，是没法正常使用的。查看了源码后，做了一些相应修改

1. 默认的DirectoryTailParserModulable类修改

他实现了2种DirectoryTailParserModulable

　　第一种是SingleLineParserModule，适用日志里只有单条记录的。并且代码中默认就是使用的这个，显然很不靠谱。提供了配置项，但是说明里没有写出来，配置项为 ‘parser’.

　第二种是MultiLineParserModule，适用多行的日志文件的。这里我们大部分情况肯定是要用这个的。

DirectoryTailSource类中如下行

  private static final String DEFAULT_PARSER_MODULE_CLASS = "com.jinoos.flume.SingleLineParserModule";

修改为。包名根据实际情况来更改

  private static final String DEFAULT_PARSER_MODULE_CLASS = "org.apache.flume.source.taildirectory.MultiLineParserModule";

2. first-line-pattern配置

这是MultiLineParserModule中的一个属性，用来验证读进来的行是否为第一行。这个说明中也没提到

如果没有配置这个配置，那么就无法正常执行，会报“wrong log format”。

主要代码如下：

 1         private void readMessage(FileSet fileSet) {
 2             try {
 3                 String buffer;
 4 
 5                 synchronized (fileSet) {
 6 
 7                     while ((buffer = fileSet.readLine()) != null) {
 8                         if (buffer.length() == 0) {
 9                             continue;
10                         }
11 
12                         boolean isFirstLine = parserModule.isFirstLine(buffer);
13                         if (isFirstLine) {
14                             sendEvent(fileSet);
15                             fileSet.appendLine(buffer);
16                             parserModule.parse(buffer, fileSet);
17 
18                         } else {
19                             if (fileSet.getLineSize() == 0) {
20                                 logger.debug("Wrong log format, " + buffer);
21                                 continue;
22                             } else {
23                                 fileSet.appendLine(buffer);
24                                 parserModule.parse(buffer, fileSet);
25                             }
26                         }
27 
28                         if (parserModule.isLastLine(buffer)) {
29                             sendEvent(fileSet);
30                         }
31                     }
32                 }
33             } catch (IOException e) {
34                 logger.warn(e.getMessage(), e);
35             }
36         }

根据我们的实际需求，我们不需要判断是否第一行，只要有change事件，全部写入到channel中即可

修改为如下方式

 1         // ??? ?? Event? ????.
 2         private void readMessage(FileSet fileSet) {
 3             try {
 4                 String buffer;
 5 
 6                 synchronized (fileSet) {
 7 
 8                     while ((buffer = fileSet.readLine()) != null) {
 9                         if (buffer.length() == 0) {
10                             continue;
11                         }
12 
13                         fileSet.appendLine(buffer);
14                         sendEvent(fileSet);
15                     }
16                 }
17             } catch (IOException e) {
18                 logger.warn(e.getMessage(), e);
19             }
20         }

改为这种方式后，只要来一行就会send到channel中。如果需要批量的，可以按自己要求更改。

现在就不再需要关注first-line-pattern这个配置了。

注意：但是配置在配置文件中还是配的，虽然它没有起到任何作用。如果想不配置，请修改MultiLineParserModule的configure（Context context）方法

3.监控文件中有中文，编码的配置添加

目前这个版本是无法支持中文的文件的。

正式读取数据的方法：位置FileSet类中

  public String readLine() throws IOException {
    return rReader.readLine();
  }

这个rReader是个RandomAccessFile对象

  public FileSet(AbstractSource source, FileObject fileObject)
      throws IOException {
    this.source = source;
    this.fileObject = fileObject;

    this.bufferList = new ArrayList<String>();

    //File f = new File(fileObject.getName().getPath());
    File f = new File("d:/tmp/log_compare/test1.txt");

    rReader = new RandomAccessFile(f, "r");
    rReader.seek(f.length());

    bufferList = new ArrayList<String>();
    headers = new HashMap<String, String>();
    logger.debug("FileSet has been created " + fileObject.getName().getPath());
    this.seq = 0L;
  }

在FileSet类实例化时创建。

下面开始修改操作，源代码中是直接使用了RandomAccessFile的readline()方法，修改为按byte读取的方式

  /**
   * 
   * @Title: readLine
   * @Description: TODO(读取文件中的一行)
   * @param @throws IOException    设定文件
   * @return String    返回类型
   * @throws
   */
  public String readLine() throws IOException {
    if(rReader.getFilePointer() < rReader.length()) {
        byte b = rReader.readByte();//读取一个byte
        int i = 0;
        byte[] buf = new byte[10240];//创建大小为1M的数据，如果你的单行超过1M，那么会出错
        //如果读到换行符，或者读到文件最后就停止。表示已经读完一行
        while(b != ‘\n‘ && rReader.getFilePointer() < rReader.length()) {
            buf[i++] = b;
            b = rReader.readByte();
            
        }
        return new String(buf,0,i);
    }else{
        return "";
    }
  }

改完后重新打包再次测试，发现已经可以支持中文了。

4.每次新文件刚被创建时会丢失第一条数据

代码如下

        public void run() {
            while (true) {
                try {
                    // DirectoryTailEvent event = eventQueue.poll(
                    // eventQueueWorkerTimeoutMiliSecond,
                    // TimeUnit.MILLISECONDS);
                    DirectoryTailEvent event = eventQueue.take();

                    if (event == null) {
                        continue;
                    }

                    if (event.type == FileEventType.FILE_CHANGED) {
                        fileChanged(event.event);
                    } else if (event.type == FileEventType.FILE_CREATED) {
                        fileCreated(event.event);
                    } else if (event.type == FileEventType.FILE_DELETED) {
                        fileDeleted(event.event);
                    } else if (event.type == FileEventType.FLUSH) {
                        if (event.fileSet != null)
                            sendEvent(event.fileSet);
                    }
                } catch (InterruptedException e) {
                    logger.debug(e.getMessage(), e);
                } catch (FileSystemException e) {
                    logger.info(e.getMessage(), e);
                }
            }
        }

上面这段代码为监测的文件夹有新的事件时的处理。这里我们要看的是FILE_CREATE事件，他调用了fileCreated(event.event);

 1     private void fileCreated(FileChangeEvent event)
 2                 throws FileSystemException {
 3             String path = event.getFile().getName().getPath();
 4             String dirPath = event.getFile().getParent().getName().getPath();
 5 
 6             logger.debug(path + " has been created.");
 7 
 8             DirPattern dirPattern = null;
 9             dirPattern = pathMap.get(dirPath);
10 
11             if (dirPattern == null) {
12                 logger.warn("Occurred create event from un-indexed directory. "
13                         + dirPath);
14                 return;
15             }
16 
17             // ???? ???? ????.
18             if (!isInFilePattern(event.getFile(), dirPattern.getFilePattern())) {
19                 logger.debug(path + " is not in file pattern.");
20                 return;
21             }
22 
23             FileSet fileSet;
24 
25             fileSet = fileSetMap.get(event.getFile().getName().getPath());
26             //fileSet = fileSetMap.get(path);
27             if (fileSet == null) {
28                 try {
29                     logger.info(path
30                             + " is not in monitoring list. It‘s going to be listed.");
31                     
32                     fileSet = new FileSet(source, event.getFile());
33                     // a little synchronized bug here.fixed by tqli,2014-08-07
34                     // ,E-mail:[email protected]
35                     synchronized (fileSetMap) {
36                         fileSetMap.put(path, fileSet);
37                     }
38                 } catch (IOException e) {
39                     logger.error(e.getMessage(), e);
40                     return;
41                 }
42             }
43         }

看第27行，当新的文件进来，需要创建一个fileSet对象。将这个fileSet对象存入fileSetMap中

看fileSet实例化的方法，上面已经贴过了

 1   public FileSet(AbstractSource source, FileObject fileObject)
 2       throws IOException {
 3     this.source = source;
 4     this.fileObject = fileObject;
 5 
 6     this.bufferList = new ArrayList<String>();
 7 
 8     File f = new File(fileObject.getName().getPath());
 9     //File f = new File("d:/tmp/log_compare/test1.txt");
10     rReader = new RandomAccessFile(f, "r");
11     rReader.seek(f.length());
12     bufferList = new ArrayList<String>();
13     headers = new HashMap<String, String>();
14     logger.debug("FileSet has been created " + fileObject.getName().getPath());
15     logger.debug("file length now is : " + f.length());
16     this.seq = 0L;
17   }

注意看第11行，将游标移到到f.length的位置，这样的问题就是跟着文件新建时写入的内容，全部被忽略了。这样就造成了数据丢失

那怎么解决这个问题呢，简单的改为

rReader.seek(0);

肯定是不行的，具体的原因，大家自己思考下吧。

我们目的的就是在有监控新的事件时，创建的fileSet，游标位置能在文件原来的的位置。
需求明确了，下面就知道该做哪些事了。
1 首先在DirectoryTailSource中start方法执行时，将配置监控文件下符合正则条件文件的length都保存在一个Map里
2 在监听到新事件新建fileSet时，判断这个文件是新建的还是之前就存在的，如果是之前就存在的，那么就可以直接取之前记下的这个文件的大小。如果不存在，说明这个文件是个新文件，则从0位置开始读
注意：这个不支持文件更改的情况，只能适应只对文件做增加的场景
下面是代码修改的部分

DirectoryTailSource类

添加 fileInitLengthMap 属性

1     private Map<String, DirPattern> dirMap;
2     private Map<String, DirPattern> pathMap;
3     private Map<String,Long> fileInitLengthMap;//文件初始大小记录，用来定位新建fileSet时的游标初始位置

在configure方法中实例化fileInitLengthMap

    public void configure(Context context) {
        logger.info("Source Configuring..");

        dirMap = new HashMap<String, DirPattern>();
        pathMap = new HashMap<String, DirPattern>();
        fileInitLengthMap = new HashMap<String,Long>();

在start方法中初始化fileInitLengthMap。保存全部符合正则条件的文件大小。红色部分为添加的代码

 1     public void start() {
 2         logger.info("Source Starting..");
 3 
 4         if (sourceCounter == null) {
 5             sourceCounter = new SourceCounter(getName());
 6         }
 7 
 8         fileSetMap = new Hashtable<String, FileSet>();
 9 
10         try {
11             fsManager = VFS.getManager();
12         } catch (FileSystemException e) {
13             logger.error(e.getMessage(), e);
14             return;
15         }
16 
17         monitorRunnable = new MonitorRunnable();
18 
19         fileMonitor = new DefaultFileMonitor(monitorRunnable);
20         fileMonitor.setRecursive(false);
21 
22         FileObject fileObject;
23 
24         logger.debug("Dirlist count " + dirMap.size());
25         for (Entry<String, DirPattern> entry : dirMap.entrySet()) {
26             logger.debug("Scan dir " + entry.getKey());
27 
28             DirPattern dirPattern = entry.getValue();
29 
30             try {
31                 fileObject = fsManager.resolveFile(dirPattern.getPath());
32             } catch (FileSystemException e) {
33                 logger.error(e.getMessage(), e);
34                 continue;
35             }
36 
37             try {
38                 if (!fileObject.isReadable()) {
39                     logger.warn("No have readable permission, "
40                             + fileObject.getURL());
41                     continue;
42                 }
43 
44                 if (FileType.FOLDER != fileObject.getType()) {
45                     logger.warn("Not a directory, " + fileObject.getURL());
46                     continue;
47                 }
48 
49                 // ??? Monitoring ??? ????.
50                 fileMonitor.addFile(fileObject);
51                 logger.debug(fileObject.getName().getPath()
52                         + " directory has been add in monitoring list");
53                 pathMap.put(fileObject.getName().getPath(), entry.getValue());
54                 //pathMap.put("d:/tmp/log_compare", entry.getValue());
55                 //新增部分，文件初始化大小保存
56                 FileObject[] allChiledfile = fileObject.getChildren();
57                 for(FileObject chiledFileobject : allChiledfile) {
58                     if(dirPattern.getFilePattern().matcher(chiledFileobject.getName().getBaseName()).find()) {
59                         String chiledFildPath = chiledFileobject.getName().getPath();
60                         //String chiledFildPath = "d:/tmp/log_compare/test1.txt";
61                         File chiledfile = new File(chiledFildPath);
62                         fileInitLengthMap.put(chiledFildPath, 
63                                 chiledfile.length());
64                         logger.debug(chiledFildPath + " init length is :" + chiledfile.length());
65                     }
66                 }
67             } catch (FileSystemException e) {
68                 logger.warn(e.getMessage(), e);
69                 continue;
70             } catch (Exception e) {
71                 logger.debug(e.getMessage(), e);
72             }
73 
74         }
75 
76         executorService = Executors
77                 .newFixedThreadPool(eventQueueWorkerSize + 1);
78         monitorFuture = executorService.submit(monitorRunnable);
79 
80         for (int i = 0; i < eventQueueWorkerSize; i++) {
81             workerFuture[i] = executorService.submit(new WorkerRunnable(this));
82         }
83 
84         sourceCounter.start();
85         super.start();
86     }

FileSet类

 1   public FileSet(AbstractSource source, FileObject fileObject,Map<String,Long> fileInitLengthMap)
 2       throws IOException {
 3     this.source = source;
 4     this.fileObject = fileObject;
 5 
 6     this.bufferList = new ArrayList<String>();
 7 
 8     File f = new File(fileObject.getName().getPath());
 9     rReader = new RandomAccessFile(f, "r");
10     /*
11      *判断在初始化taildirSource时，这个文件是否存在，如果存在则游标定位当时记录下的文件长度开始
12      *如果不存在，则说明这是一个新建的文件，游标从0开始
13      */
14     if(fileInitLengthMap.containsKey(fileObject.getName().getPath())) {
15         rReader.seek(fileInitLengthMap.get(fileObject.getName().getPath()));
16     }else{
17         rReader.seek(0);
18     }
19     
20     bufferList = new ArrayList<String>();
21     headers = new HashMap<String, String>();
22     logger.debug("FileSet has been created " + fileObject.getName().getPath());
23     logger.debug("file length now is : " + f.length());
24     this.seq = 0L;
25   }

修改类实例化的方法。并修改DirectoryTailSource类中调用FileSet实例化方法的地方。

至此修改全部全部完成。

没找到能上传附件的地方，改完的jar包就不提供了。

此为一个使用这个jar的例子

a.sources = sources
a.sinks = sinks
a.channels = c

#configure sources
a.sources.sources.type = org.apache.flume.source.taildirectory.DirectoryTailSource
a.sources.sources.dirs = s0
#a.sources.sources.dirs.s0.path = /usr/local/nginx/logs/
a.sources.sources.dirs.s0.path = /home/flume/testTailDir
a.sources.sources.dirs.s0.file-pattern = ^access_.*log$
a.sources.sources.first-line-pattern = ^(.*)$
#congfigure sinks
a.sinks.sinks.type = file_roll
a.sinks.sinks.sink.directory = /home/flume/testTailDir2
a.sinks.sinks.sink.rollInterval = 30
a.sinks.sinks.channel = c
#configure channals
a.channels.c.type = memory
#bind channel
a.sources.sources.channels = c