《ELK日志采集监控方案》由会员分享，可在线阅读，更多相关《ELK日志采集监控方案（10页珍藏版）》请在金锄头文库上搜索。
1、ELK日志采集监控方案概述基于ELK搭建日志实时采集监控系统系统要求优化系统内核配置：1. ulimit2. logstash各组件间有版本依赖，请注意使用的版本匹配3. 配置好邮件服务sendmail系统组成Elasticsearch +Logstash+ Kibana（ELK）：1. Elasticsearch：日志汇总存储并创建索引提供查询服务2. Logstash：日志采集、过滤与监控，logstash本身提供多种input-output策略，可灵活选择3. Kibana：提供统一的日志查询web界面，同时可以进行分类图表统计架构策略在应用服务端配置logstash-agent负责采集
2、原始日志并汇总到MQ，MQ选用kafka进行日志消息缓存与分发，后端部署logstash-indexer，订阅kafka topic内日志消息并写入ES文件存储，同时判断日志消息内是否包含错误或异常等信息，通过sendMail邮件服务发送报警邮件到相关邮件组，后端ES部署双节点集群做分布式搜索服务，提供http服务给kibana供用户通过web界面查询实时日志。如图：1 JDK安装略2 ES注：l ES最新版本要求不能以root方式启动，需创建对应用户并赋予es相关文件路径的控制权限l ES2.0版与kibana新版整合时，有js兼容问题，kibana UI界面在IE、chrome下无法打开，
3、Firefox可以2.1单节点模式1. 下载es tar官网：wget 2. 解压并配置tar elasticsearch-1.7.0.tar.gz修改$es_home/ config/ elasticsearch.yml注：红色为基础配置项，需针对实际场景修改，更多详细配置请参阅官网#配置es集群命名:#当前节点名#数据存储目录，默认$es_home/datapath.data #日志目录，默认$es_home/logspath.logs#临时文件存储目录，默认$es_home/workpath.work#对外服务http端口，默认9200http.port#节点间交互端口，默认9300tr
4、ansport.tcp.port3. 使用启动：bin/elasticsearch后端启动：bin/elasticsearch dPID启动：bin/elasticsearch -d -p pid4. 以服务方式管理es：对elasticsearch执行命令的包装服务，安装之后，方便elasticsearch的启动，停止等等操作老版本es适用：elasticsearchservicewrapper:（1）下载elasticsearchservicewrapper:git clone然后将目录下的service目录拷贝至ES_HOME/bin目录下（2）简单配置jvm的内存修改ES_HOME/b
5、in/service/elasticsearch.conf，set.default.ES_HEAP_SIZE=1024，该值根据机器的配置可自定义。（3）安装启动服务执行命令：ES_HOME/bin/service/elasticsearch install（4）启动/停止/重启服务执行命令：ES_HOME/bin/service/elasticsearch start/stop/restart新版本es适用参见：5. Web请求，验证服务是否成功返回：6. Web查询数据接口：7. Web查询集群信息接口：8.9.10.2.2集群模式在其他节点按上述单节点模式安装，修改配置文件$es_hom
6、e/ config/ elasticsearch.yml:保证：1. 各节点：必须一致：必须不同2. 各节点在相同网段内启动各节点后，es会自动发现同网段内的节点组成集群2.3管理插件1. head：查看集群几乎所有信息，还能进行简单的搜索查询，观察自动恢复的情况等安装：./bin/plugin -install mobz/elasticsearch-head查看：2. Bigdesk：集群监控插件，通过该插件可以查看整个集群的资源消耗情况，cpu、内存、http链接等安装：./bin/plugin -install lukas-vlcek/bigdesk查看：3. Marvel: 商业集群监
7、控插件,同时可以辅助curator实现定时自动删除旧索引数据,更多高级功能需购买license4. 插件大全：3 Logstash下载：Logstash-input-plugins使用文档:Logstash-output-plugins使用文档:解压logstash-all-plugins-2.1.0.tar.gz，配置启动文件，进行不同模式处理。3.1 logstash-agent功能：监控、过滤日志,也可称为shipper配置日志采集，采集原始日志发送到kafka集群logstash-agent.conf：input # file path = $log_path#log typetype
8、 = “$log_type” output kafka #kafka brokers bootstrap_servers = $borker_list#send to kafka topic topic_id = $topic_name codec = plain format = %message #stdout codec = rubydebug 注：1. output使用kafka时，codec默认采用json格式发送，如显式配置了codec为plain时不会传输“version，type，tags”等input内的默认字段和add_field添加的字段，可通过format自定义传输格式
9、：如：codec 为json时传输：message:dfsd,version:1,timestamp:2015-12-07T07:18:01.124Z,type:in,IP:%ip,host:testqiuli.novalocal,tags:_grokparsefailurecodec为plain时传输：2015-12-07T07:22:21.526Z testqiuli.novalocal tt2.3.2 logstash-indexer功能：收集日志并将日志交给ElasticSearch 做搜索配置统一订阅kafka topic中日志，收集汇总到eslogstash-index.conf：
10、input #kafka-input kafka zk_connect = $zk_nodestype = “$log_type”#consumer thread num,must be less then the partitions of kafka topicconsumer_threads = 2#topic_id = $topic_name#kafka topics to consume,more topics such as :topic1,topic2 white_list = $topic_names output elasticsearch hosts = $es_IP:$e
11、s_port 3.3 报警filter配置filter，对日志中包含“error”或“exception”等关键字的异常错误进行过滤，并通过邮件报警到指定的邮件组注：报警的output必须在output模块最后添加，否则匹配上的日志信息不会由其他output收集了配置logstash-warn.conf：input #kafka-input kafka zk_connect = $zk_nodestype = “$log_type”#topic_id = $topic_name#kafka topics to consume,more topics such as :topic1,topic2
12、 white_list = $topic_names #warning filterfilter grok match = message = error, exception add_tag = tag_error output elasticsearch hosts = $es_IP:$es_port #warning output,ps:make sure at the end of output if tag_error in tags exec command = echo %timestamp %source: %message | mail -s $email_title $em
13、ail_to” 3.4 多个input-output同一个conf文件中可配置多个input和output,完成不同功能logstash-all.conf：input # file path = $log_pathtype = “$log_type” #kafka-input kafka zk_connect = $zk_nodes#topic_id = $topic_name#kafka topics to consume,more topics such as :topic1,topic2 white_list = $topic_names filter grok match = message = error, exception add_tag = tag_error output elasticsearch hosts = $es_IP:$es_port kafka #kafka brokers bootstrap_servers = $borker_list#send to kafka topic topic_id = $topic_name
