實時收集Storm日誌到ELK集羣

時間 2019-11-12

標籤實時收集 storm 日誌 elk 集羣欄目 Storm 简体版

原文原文鏈接

背景

咱們的storm實時流計算項目已經上線幾個月了，因爲各類緣由遲遲沒有進行監控，每次出現問題都要登陸好幾臺機器，而後使用sed，shell，awk，vi等各類命令來查詢緣由，效率很是低下，並且有些統計是無法作的，因此頗有必要對storm自己相關的日誌以及咱們運行在storm上面的任務的日誌作一個統一的日誌收集，分析，查詢，統計平臺。shell

技術選型

對於這個選擇，其實不用考慮那麼多,借用一句名言 Life is short , You need elk ! 關於elk相關的安裝這裏再也不介紹，可參考散仙的博客： http://qindongliang.iteye.com/category/330375ruby

需求分析

序號	討論	內容
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定義app

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的根目錄下面elasticsearch

啓動腳本：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`