实时收集Storm日志到ELK集群

我会带着你远行 2022-04-11 09:08 179阅读 0赞

\#\#\#\# 背景  
  
我们的storm实时流计算项目已经上线几个月了，由于各种原因迟迟没有进行监控，每次出现问题都要登录好几台机器，然后使用sed，shell，awk，vi等各种命令来查询原因，效率非常低下，而且有些统计是没法做的，所以很有必要对storm本身相关的日志以及我们运行在storm上面的任务的日志做一个统一的日志收集，分析，查询，统计平台。  
  
  
\#\#\#\# 技术选型  
  
对于这个选择，其实不用考虑那么多,借用一句名言 Life is short , You need elk !  
关于elk相关的安装这里不再介绍，可参考散仙的博客：  
http://qindongliang.iteye.com/category/330375  
  
  
\#\#\#\# 需求分析  
  
  
序号|讨论 | 内容  
|---|---|---  
1|storm需要收集的日志 | （1）本身的相关的日志 （2）提交任务的日志  
2|日志过滤| 排除gc的log和部分不相干业务的log  
3|索引如何切分| 考虑量不是很大，按每月生成一份索引  
4|索引模板定制|默认的动态mapping比较简答，所以我们采用自定义动态索引模板  
5|日志的定期删除|使用es官网插件curator管理  
  
  
\#\#\#\# 核心配置  
  
（1）es的模板定义  
注意date类型的动态类型是开启docvalue的，便于快速聚合和排序

{
       
      "order": 0,
      "template": "jstorm*",
      "settings": {
       
        "index": {
       
          "number_of_replicas": "0",
          "number_of_shards": "3"
        }
      },
      "mappings": {
       
        "_default_": {
       
          "dynamic_templates": [
            {
       
              "level": {
       
                "mapping": {
       
                  "index": "not_analyzed",
                  "type": "string"
                },
                "match": "level",
                "match_mapping_type": "string"
              }
            },
            {
       
              "message": {
       
                "mapping": {
       
                  "index": "analyzed",
                  "type": "string"
                },
                "match": "message",
                "match_mapping_type": "string"
              }
            },
            {
       
              "date_fields": {
       
                "mapping": {
       
                  "doc_values": true,
                  "type": "date"
                },
                "match_mapping_type": "date"
              }
            },
            {
       
              "string_fields": {
       
                "mapping": {
       
                  "index": "not_analyzed",
                  "type": "string"
                },
                "match": "*",
                "match_mapping_type": "string"
              }
            }
          ],
          "_all": {
       
            "enabled": false
          }
        }
      },
      "aliases": {}
    }

（2）logstash的conf定义

input{
       
        file{
       
                #初始化全量导入
                start_position => "beginning"    
                #统一的storm的日志目录
                path=> ["/data/logs/jstorm/**/*.log"]   
                #排除的路径
                exclude =>["*gc*","*log_monitor*"]
                #指定文件偏移量存储的文件
                sincedb_path => "./sincedb" 
                #配置多行数据收集（针对异常）
                codec => multiline {
       
                              #类似两个info之间的所有数据是一行数据    
                              pattern => "^\[%{LOGLEVEL:loglevel}"
                              #true代表是两个loglevel之间的数据
                              #false代表两个异常之间的数据,跟上面的相反
                              negate=> true
                              #后一条的数据前面所有的，都属于这整条数据
                              what => "previous"
                            }
            }
    } 
    
    
    filter {
       
            #使用gork直接获取日志级别和时间
            grok {
       
                    match =>{"message"=>"%{LOGLEVEL:loglevel}\s*%{TIMESTAMP_ISO8601:time} "}
            }
    
        #  转化日志时间为收集的时间，并移除无用的字段
        date{     
                match => ["time","yyyy-MM-dd HH:mm:ss.SSS","yyyy-MM-dd HH:mm:ss","ISO8601"]      
                remove_field => [ "time","@version" ]   
    
       } 
    
    # 这个地方可以对一些数据做过滤
    #  if [loglevel] == "DEBUG" {
       
    #   drop { }
    #  }
    
    }
    
    
    #输出到es的配置
    output{
       
    
      elasticsearch{ 
       #设置索引名
       index => "jstorm_pro%{+YYYY-MM}"   
       hosts=> ["192.168.8.5:9200","192.168.8.6:9200","192.168.8.7:9200"]  
       #关闭logstash自动管理模块
       manage_template => false
       #指定模板名为jstrom
       template_name => "jstorm"  
       #设置flush的数量
       flush_size => 3000  
       }
     # 调试控制台输出    
     # stdout { codec => rubydebug  }
    }

\#\#\#\# 辅助脚本  
放在logstash的根目录下面

启动脚本：start_jstorm.sh
    nohup bin/logstash -f config/jstorm.conf  &> jstorm_logstash.log & echo $! >jstorm_logstash_pid& 
    
    关闭脚本：stop_jstorm.sh
    kill -9 `cat jstorm_logstash_pid`

\#\#\#\# 收集检索效果  
一切完成后，启动logstash收集进程后，我们就可以实时在kibana里面分析数据了，非常nice！  
\[img\]http://note.youdao.com/yws/public/resource/344cd03f173c19ea03136b36d01a5f91/xmlnote/7481853568384AADAE4A3FD7308FFBDB/25321\[/img\]  
  
然后，我们就可以非常快速的定位异常数据了。  
  
\[b\]\[color=green\]\[size=large\]  
有什么问题可以扫码关注微信公众号：我是攻城师(woshigcs)，在后台留言咨询。  
技术债不能欠，健康债更不能欠， 求道之路，与君同行。  
\[/size\]\[/color\]\[/b\]  
\[img\]http://dl2.iteye.com/upload/attachment/0104/9948/3214000f-5633-3c17-a3d7-83ebda9aebff.jpg\[/img\]