图文精华

About云-梭伦科技»专题 › 技术学习(版主发帖区) › 大数据学习 › Flume › 连载型› Flume-ng生产环境实践（四）实现log格式化interceptor
分享

Flume-ng生产环境实践（四）实现log格式化interceptor

坎蒂丝_Swan 2015-1-12 12:09:44 发表于 连载型 [显示全部楼层] 回帖奖励 阅读模式 关闭右栏 3 31948
About云VIP帮助找工作和提升
本帖最后由 坎蒂丝_Swan 于 2015-1-12 14:23 编辑
问题导读
1.filesink中数据如何传输？
2.event中header的键值对是怎样起作用的？











续上篇，由于filesink中需要使用/data/log/%{dayStr}/log-%{hourStr}%{minStr}-这样文件格式的，为了使file-sink能使用%{dayStr}这样的标签，需要在数据传输过程中，给event的header中添加对应的键值对。在flume-ng中提供了很方便的方式：Interceptor以下为实现的interceptor，首先使用正则表达式匹配nginx日志，如何匹配成功，则获取匹配到的数据，并且对url中的参数进行处理，最后所有日志信息都被存储在Map中。根据配置文件中需要输出的键找到对应的值，按照顺序输出为csv格式的行。

原始日志格式： 
  1. 112.245.239.72 - - [29/Dec/2012:15:00:00 +0800] "GET /p.gif?a=1&b=2 HTTP/1.1" 200 0 "Mozilla/4.0 (compatible; MSIE 8.0; Windows NT 5.1; Trident/4

  2. .0; 4399Box.1357; .NET CLR 2.0.50727; .NET CLR 3.0.4506.2152; .NET CLR 3.5.30729; AskTbPTV2/5.9.1.14019; 4399Box.1357)"
复制代码

最终结果：
  1. 1,2
复制代码

配置信息为：
  1. agent.sources = source

  2. agent.channels = channel

  3. agent.sinks = sink

  4. 

  5. agent.sources.source.type = exec

  6. #agent.sources.source.command = tail -n +0 -F /data/tmp/accesspvpb_2012-11-18.log

  7. agent.sources.source.command = cat /opt/nginx/logs/vvaccess_log_pipe

  8. agent.sources.source.interceptors = logformat

  9. 

  10. agent.sources.source.interceptors.logformat.type = org.apache.flume.interceptor.LogFormatInterceptor$Builder

  11. agent.sources.source.interceptors.logformat.confpath = /usr/programs/flume/conf/logformat_vv.properties

  12. agent.sources.source.interceptors.logformat.dynamicprop = true

  13. agent.sources.source.interceptors.logformat.hostname = vv111

  14. agent.sources.source.interceptors.logformat.prop.monitor.rollInterval = 100000

  15. # The channel can be defined as follows.

  16. agent.sources.source.channels = channel

  17. 

  18. 

  19. agent.sinks.sink.type = avro

  20. agent.sinks.sink.hostname = 192.168.0.100

  21. agent.sinks.sink.port = 44444

  22. agent.sinks.sink.channel = channel

  23. 

  24. # Each channel's type is defined.

  25. agent.channels.channel.type = file

  26. agent.channels.channel.checkpointDir = /data/tmpc/checkpoint

  27. agent.channels.channel.dataDirs = /data/tmpc/data

  28. agent.channels.channel.transactionCapacity = 15000
复制代码

/usr/programs/flume/conf/logformat_vv.properties文件内容为：
  1. keys=a,b

  2. regexp=([0-9]{1,3}\\.[0-9]{1,3}\\.[0-9]{1,3}\\.[0-9]{1,3})\\s-\\s-\\s\\[([^]]+)\\]\\s"GET\\s/p.gif\\?(.+)\\s.*"\\s[0-9]+\\s[0-9]+\\s"(.+)"
复制代码

interceptor的代码：
  1. package org.apache.flume.interceptor;

  2. 

  3. import static org.apache.flume.interceptor.LogFormatInterceptor.Constants.CONF_PATH;

  4. import static org.apache.flume.interceptor.LogFormatInterceptor.Constants.DYNAMICPROP;

  5. import static org.apache.flume.interceptor.LogFormatInterceptor.Constants.DYNAMICPROP_DFLT;

  6. import static org.apache.flume.interceptor.LogFormatInterceptor.Constants.HOSTNAME;

  7. import static org.apache.flume.interceptor.LogFormatInterceptor.Constants.HOSTNAME_DFLT;

  8. import static org.apache.flume.interceptor.LogFormatInterceptor.Constants.PROPMONITORINTERVAL;

  9. import static org.apache.flume.interceptor.LogFormatInterceptor.Constants.PROPMONITORINTERVAL_DFLT;

  10. 

  11. import java.io.File;

  12. import java.io.FileInputStream;

  13. import java.io.FileNotFoundException;

  14. import java.io.IOException;

  15. import java.text.ParseException;

  16. import java.text.SimpleDateFormat;

  17. import java.util.Date;

  18. import java.util.HashMap;

  19. import java.util.LinkedList;

  20. import java.util.List;

  21. import java.util.Map;

  22. import java.util.Properties;

  23. 

  24. import org.apache.flume.Context;

  25. import org.apache.flume.Event;

  26. import org.apache.flume.event.EventBuilder;

  27. import org.apache.oro.text.regex.MalformedPatternException;

  28. import org.apache.oro.text.regex.MatchResult;

  29. import org.apache.oro.text.regex.Pattern;

  30. import org.apache.oro.text.regex.PatternCompiler;

  31. import org.apache.oro.text.regex.PatternMatcher;

  32. import org.apache.oro.text.regex.Perl5Compiler;

  33. import org.apache.oro.text.regex.Perl5Matcher;

  34. import org.slf4j.Logger;

  35. import org.slf4j.LoggerFactory;

  36. 

  37. public class LogFormatInterceptor implements Interceptor {

  38. 

  39.         private static final Logger logger = LoggerFactory

  40.                      . getLogger(LogFormatInterceptor.class);

  41. 

  42.         private String conf_path = null;

  43.         private boolean dynamicProp = false;

  44.         private String hostname = null;

  45. 

  46.         private long propLastModify = 0;

  47.         private long propMonitorInterval ;

  48. 

  49.         private String regexp = null;

  50.         private List<String> keys = null;

  51. 

  52.         private Pattern pattern = null;

  53.         private PatternCompiler compiler = null;

  54.         private PatternMatcher matcher = null;

  55.         private SimpleDateFormat sdf = null;

  56.         private SimpleDateFormat sd = null;

  57.         private SimpleDateFormat sh = null;

  58.         private SimpleDateFormat sm = null;

  59.         private SimpleDateFormat sdfAll = null;

  60. 

  61.         private long eventCount = 0l;

  62. 

  63.         public LogFormatInterceptor(String conf_path, boolean dynamicProp,

  64.                      String hostname, long propMonitorInterval) {

  65.                this.conf_path = conf_path;

  66.                this.dynamicProp = dynamicProp;

  67.                this.hostname = hostname;

  68.                this.propMonitorInterval = propMonitorInterval;

  69.        }

  70. 

  71.         @Override

  72.         public void close () {

  73. 

  74.        }

  75. 

  76.         @Override

  77.         public void initialize () {

  78.                try {

  79.                       // 读取配置文件，初始化正在表达式和输出的key列表

  80.                      File file = new File(conf_path );

  81.                       propLastModify = file.lastModified();

  82.                      Properties props = new Properties();

  83.                      FileInputStream fis;

  84.                      fis = new FileInputStream(file);

  85.                      props.load(fis);

  86.                       regexp = props.getProperty( "regexp");

  87.                      String strKey = props.getProperty( "keys");

  88.                       if (strKey != null) {

  89.                            String[] strkeys = strKey.split( ",");

  90.                             keys = new LinkedList<String>();

  91.                             for (String key : strkeys) {

  92.                                    keys.add(key);

  93.                            }

  94.                      }

  95.                       if (keys == null) {

  96.                             logger.error("====================keys is null====================");

  97.                      } else {

  98.                             logger.info("keys=" + keys );

  99.                      }

  100.                       if (regexp == null) {

  101.                             logger.error("====================regexp is null====================");

  102.                      } else {

  103.                             logger.info("regexp=" + regexp );

  104.                      }

  105. 

  106.                       // 初始化正在表达式以及时间格式化类

  107.                       compiler = new Perl5Compiler();

  108.                       pattern = compiler.compile( regexp);

  109.                       matcher = new Perl5Matcher();

  110. 

  111.                       sdf = new SimpleDateFormat("dd/MMM/yyyy:HH:mm:ss Z" ,

  112.                                   java.util.Locale. US);

  113.                       sd = new SimpleDateFormat("yyyyMMdd" );

  114.                       sh = new SimpleDateFormat("HH" );

  115.                       sm = new SimpleDateFormat("mm" );

  116.                       sdfAll = new SimpleDateFormat("yyyyMMddHHmmss" );

  117. 

  118.               } catch (MalformedPatternException e) {

  119.                       logger.error("Could not complile pattern!" , e);

  120.               } catch (FileNotFoundException e) {

  121.                       logger.error("conf file is not found!" , e);

  122.               } catch (IOException e) {

  123.                       logger.error("conf file can not be read!" , e);

  124.               }

  125.        }

  126. 

  127.         @Override

  128.         public Event intercept(Event event) {

  129.               ++ eventCount;

  130.                try {

  131.                       if (dynamicProp && eventCount > propMonitorInterval) {

  132.                            File file = new File(conf_path );

  133.                             if (file.lastModified() > propLastModify ) {

  134.                                    propLastModify = file.lastModified();

  135.                                   Properties props = new Properties();

  136.                                   FileInputStream fis;

  137.                                   fis = new FileInputStream(file);

  138.                                   props.load(fis);

  139.                                   String strKey = props.getProperty( "keys");

  140.                                    if (strKey != null) {

  141.                                          String[] strkeys = strKey.split("," );

  142.                                          List<String> keystmp = new LinkedList<String>();

  143.                                           for (String key : strkeys) {

  144.                                                 keystmp.add(key);

  145.                                          }

  146.                                           if (keystmp.size() > keys .size()) {

  147.                                                  keys = keystmp;

  148.                                                  logger.info("dynamicProp status updated = " + keys);

  149.                                          } else {

  150.                                                  logger.error("dynamicProp status new keys size less than old,so status update fail = "

  151.                                                               + keys);

  152.                                          }

  153.                                   } else {

  154.                                           logger.error("dynamicProp status get keys fail ,so status update fail = "

  155.                                                        + keys);

  156.                                   }

  157. 

  158.                            }

  159.                      }

  160. 

  161.                      Map<String, String> headers = event.getHeaders();

  162.                      headers.put( "host", hostname );

  163.                      String body = new String(event.getBody());

  164.                       if (pattern != null) {

  165.                            StringBuffer stringBuffer = new StringBuffer();

  166.                            Date date = null;

  167.                            Map<String, String> index = new HashMap<String, String>();

  168.                             if (matcher .contains(body, pattern)) {

  169.                                   index.put( "host", hostname );

  170.                                   MatchResult result = matcher.getMatch();

  171.                                   index.put( "ip", result.group(1));

  172.                                    try {

  173.                                          date = sdf.parse(result.group(2));

  174.                                          index.put( "loc_time", sdfAll.format(date));

  175.                                   } catch (ParseException e1) {

  176. 

  177.                                   }

  178.                                   String url = result.group(3).replaceAll( ",", "|");

  179.                                   String[] params = url.split( "&");

  180.                                    for (String param : params) {

  181.                                          String[] p = param.split("=" );

  182.                                           if (p.length == 2) {

  183.                                                 index.put(p[0], p[1]);

  184.                                          }

  185.                                   }

  186.                                   index.put( "browser", result.group(4).replaceAll("," , "|" ));

  187.                                    for (String key : keys ) {

  188.                                           if (index.containsKey(key)) {

  189.                                                 stringBuffer.append(index.get(key) + ",");

  190.                                          } else {

  191.                                                 stringBuffer.append( "~,");

  192.                                          }

  193.                                   }

  194.                                    if (stringBuffer.length() > 0) {

  195.                                          stringBuffer.deleteCharAt(stringBuffer.length() - 1);

  196.                                   } else {

  197.                                          stringBuffer.append( "error=" + body);

  198.                                   }

  199. 

  200.                                    if (date != null) {

  201.                                          headers.put( "dayStr", sd .format(date));

  202.                                          headers.put( "hourStr", sh .format(date));

  203.                                          Integer m = Integer.parseInt(sm.format(date));

  204.                                          String min = "";

  205.                                           if (m >= 0 && m < 10) {

  206.                                                 min = "0" + (m / 5) * 5;

  207.                                          } else {

  208.                                                 min = (m / 5) * 5 + "" ;

  209.                                          }

  210.                                          headers.put( "minStr", min);

  211.                                   } else {

  212.                                          headers.put( "dayStr", "errorLog" );

  213.                                   }

  214.                                   Event e = EventBuilder.withBody(stringBuffer.toString()

  215.                                                 .getBytes(), headers);

  216.                                    return e;

  217.                            }

  218.                      }

  219.               } catch (Exception e) {

  220.                       logger.error("LogFormat error!" , e);

  221.               }

  222.                return null ;

  223.        }

  224. 

  225.         @Override

  226.         public List<Event> intercept(List<Event> events) {

  227.               List<Event> list = new LinkedList<Event>();

  228.                for (Event event : events) {

  229.                      Event e = intercept(event);

  230.                       if (e != null ) {

  231.                            list.add(e);

  232.                      }

  233.               }

  234.                return list;

  235.        }

  236. 

  237.         /**

  238.         * Builder which builds new instances of the HostInterceptor.

  239.         */

  240.         public static class Builder implements Interceptor.Builder {

  241. 

  242.                private String confPath ;

  243.                private boolean dynamicProp ;

  244.                private String hostname ;

  245.                private long propMonitorInterval ;

  246. 

  247.                @Override

  248.                public Interceptor build() {

  249.                       return new LogFormatInterceptor(confPath, dynamicProp, hostname,

  250.                                    propMonitorInterval);

  251.               }

  252. 

  253.                @Override

  254.                public void configure(Context context) {

  255.                       confPath = context.getString( CONF_PATH);

  256.                       dynamicProp = context.getBoolean(DYNAMICPROP, DYNAMICPROP_DFLT);

  257.                       hostname = context.getString( HOSTNAME, HOSTNAME_DFLT );

  258.                       propMonitorInterval = context.getLong(PROPMONITORINTERVAL,

  259.                                    PROPMONITORINTERVAL_DFLT);

  260.               }

  261. 

  262.        }

  263. 

  264.         public static class Constants {

  265. 

  266.                public static String CONF_PATH = "confpath";

  267. 

  268.                public static String DYNAMICPROP = "dynamicprop";

  269.                public static boolean DYNAMICPROP_DFLT = false;

  270. 

  271.                public static String HOSTNAME = "hostname";

  272.                public static String HOSTNAME_DFLT = "hostname";

  273. 

  274.                public static String PROPMONITORINTERVAL = "prop.monitor.rollInterval" ;

  275.                public static long PROPMONITORINTERVAL_DFLT = 500000l;

  276. 

  277.        }

  278. 

  279. }
复制代码




Flume-ng生产环境实践（一）Flume-ng生产环境编译
Flume-ng生产环境实践（二）flume-ng 测试过程中event丢失部分body数据
Flume-ng生产环境实践（三）实现文件sink，按照固定格式目录输出
Flume-ng生产环境实践（四）实现log格式化interceptor

欢迎加入about云群90371779 、322273151、432264021 ，云计算爱好者群，亦可关注about云腾讯认证空间||关注本站微信
回复

使用道具 举报

自定义广告语

已有(3)人评论

跳转到指定楼层
正序浏览
stark_summer 发表于 2015-1-12 13:56:28
回复

使用道具 举报

355815741 发表于 2015-1-12 22:26:29
学习了，谢谢分享~
回复

使用道具 举报

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

发表新帖
坎蒂丝_Swan

高级会员

关注

  • 148

    主题

  • 148

    帖子

  • 52

    粉丝

TA的主题
关闭

推荐上一条 /2 下一条

关于我们· 联系我们· 加入我们· 小黑屋· 合作伙伴   

Copyright © 2001-2024 About云-梭伦科技 Powered by Discuz! X3.4 Licensed Discuz Team.

简书 / Smrz 京ICP备2020039040号 Wxb 简书网举报电话：021-34700000