实时收集Storm日志到ELK集群

背景

我们的storm实时流计算项目已经上线几个月了，由于各种原因迟迟没有进行监控，每次出现问题都要登录好几台机器，然后使用sed，shell，awk，vi等各种命令来查询原因，效率非常低下，而且有些统计是没法做的，所以很有必要对storm本身相关的日志以及我们运行在storm上面的任务的日志做一个统一的日志收集，分析，查询，统计平台。

技术选型

对于这个选择，其实不用考虑那么多,借用一句名言 Life is short , You need elk ! 关于elk相关的安装这里不再介绍，可参考散仙的博客：http://qindongliang.iteye.com/category/330375

需求分析

核心配置

（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！

然后，我们就可以非常快速的定位异常数据了。