ELK日志采集监控方案

ELK日志采集监控方案概述基于ELK搭建日志实时采集监控系统系统要求优化系统内核配置：1.ulimit2.logstash各组件间有版本依赖，请注意使用的版本匹配3.配置好邮件服务sendmail系统组成Elasticsearch+Logstash+Kibana（ELK）：1.Elasticsearch：日志汇总存储并创建索引提供查询服务2.Logstash：日志采集、过滤与监控，logstash本身提供多种input-output策略，可灵活选择3.Kibana：提供统一的日志查询web界面，同时可以进行分类图表统计架构策略在应用服务端配置logstash-agent负责采集原始日志并汇总到MQ，MQ选用kafka进行日志消息缓存与分发，后端部署logstash-indexer，订阅kafkatopic内日志消息并写入ES文件存储，同时判断日志消息内是否包含错误或异常等信息，通过sendMail邮件服务发送报警邮件到相关邮件组，后端ES部署双节点集群做分布式搜索服务，提供http服务给kibana供用户通过web界面查询实时日志。如图：1JDK安装略2ES注：ES最新版本要求不能以root方式启动，需创建对应用户并赋予es相关文件路径的控制权限ES2.0版与kibana新版整合时，有js兼容问题，kibanaUI界面在IE、chrome下无法打开，Firefox可以2.1单节点模式1.下载estar官网：解压并配置tarelasticsearch-1.7.0.tar.gz修改${es_home}/config/elasticsearch.yml注：红色为基础配置项，需针对实际场景修改，更多详细配置请参阅官网#配置es集群命名cluster.name:#当前节点名node.name#数据存储目录，默认${es_home}/datapath.data#日志目录，默认${es_home}/logspath.logs#临时文件存储目录，默认${es_home}/workpath.work#对外服务http端口，默认9200http.port#节点间交互端口，默认9300transport.tcp.port3.使用启动：bin/elasticsearch后端启动：bin/elasticsearch–dPID启动：bin/elasticsearch-d-ppid4.以服务方式管理es：对elasticsearch执行命令的包装服务，安装之后，方便elasticsearch的启动，停止等等操作老版本es适用：elasticsearchservicewrapper:（1）下载elasticsearchservicewrapper:gitclone然后将目录下的service目录拷贝至ES_HOME/bin目录下（2）简单配置jvm的内存修改ES_HOME/bin/service/elasticsearch.conf，set.default.ES_HEAP_SIZE=1024，该值根据机器的配置可自定义。（3）安装启动服务执行命令：ES_HOME/bin/service/elasticsearchinstall（4）启动/停止/重启服务执行命令：ES_HOME/bin/service/elasticsearchstart/stop/restart新版本es适用参见：请求，验证服务是否成功返回：6.Web查询数据接口：查询集群信息接口：集群模式在其他节点按上述单节点模式安装，修改配置文件${es_home}/config/elasticsearch.yml:保证：1.各节点cluster.name：必须一致node.name：必须不同2.各节点在相同网段内启动各节点后，es会自动发现同网段内的节点组成集群2.3管理插件1.head：查看集群几乎所有信息，还能进行简单的搜索查询，观察自动恢复的情况等安装：./bin/plugin-installmobz/elasticsearch-head查看：：集群监控插件，通过该插件可以查看整个集群的资源消耗情况，cpu、内存、http链接等安装：./bin/plugin-installlukas-vlcek/bigdesk查看：:商业集群监控插件,同时可以辅助curator实现定时自动删除旧索引数据,更多高级功能需购买license插件大全：下载：使用文档:使用文档:解压logstash-all-plugins-2.1.0.tar.gz，配置启动文件，进行不同模式处理。3.1logstash-agent功能：监控、过滤日志,也可称为shipper配置日志采集，采集原始日志发送到kafka集群logstash-agent.conf：input{#file-inputfile{path=${log_path}#logtypetype=“${log_type}”}}output{kafka{#kafkabrokersbootstrap_servers=${borker_list}#sendtokafkatopictopic_id=${topic_name}codec=plain{format=%{message}}}#stdout{codec=rubydebug}}注：1.output使用kafka时，codec默认采用json格式发送，如显式配置了codec为plain时不会传输“@version，type，tags”等input内的默认字段和add_field添加的字段，可通过format自定义传输格式：如：codec为json时传输：{message:dfsd,@version:1,@timestamp:2015-12-07T07:18:01.124Z,type:in,IP:%{[ip]},host:testqiuli.novalocal,tags:[_grokparsefailure]}codec为plain时传输：2015-12-07T07:22:21.526Ztestqiuli.novalocaltt2.3.2logstash-indexer功能：收集日志并将日志交给ElasticSearch做搜索配置统一订阅kafkatopic中日志，收集汇总到eslogstash-index.conf：input{#kafka-inputkafka{zk_connect=${zk_nodes}type=“${log_type}”#consumerthreadnum,mustbelessthenthepartitionsofkafkatopicconsumer_threads=2#topic_id=${topic_name}#kafkatopicstoconsume,moretopicssuchas:topic1,topic2white_list=${topic_names}}}output{elasticsearch{hosts=[${es_IP}:${es_port}]}}3.3报警filter配置filter，对日志中包含“error”或“exception”等关键字的异常错误进行过滤，并通过邮件报警到指定的邮件组注：报警的output必须在output模块最后添加，否则匹配上的日志信息不会由其他output收集了配置logstash-warn.conf：input{#kafka-inputkafka{zk_connect=${zk_nodes}type=“${log_type}”#topic_id=${topic_name}#kafkatopicstoconsume,moretopicssuchas:topic1,topic2white_list=${topic_names}}}#warningfilterfilter{grok{match={message=[error,exception]}add_tag=[tag_error]}}output{elasticsearch{hosts=[${es_IP}:${es_port}]}#warningoutput,ps:makesureattheendofoutputiftag_errorin[tags]{exec{command=echo'%{@timestamp}%{source}:%{message}'|mail-s${email_title}${email_to}”}}}3.4多个input-output同一个conf文件中可配置多个input和output,完成不同功能logstash-all.conf：input{#file-inputfile{path=${log_path}type=“${log_type}”}#kafka-inputkafka{zk_connect=${zk_nodes}#topic_id=${topic_name}#kafkatopicstoconsume,moretopicssuchas:topic1,topic2white_list=${topic_names}}}filter{grok{match={message=[error,exception]}add_tag=[tag_error]}}output{elasticsearch{hosts=[${es_IP}:${es_port}]}kafka{#kafkabrokersbootstrap_servers=${borker_list}#sendtokafkatopictopic_id=${topic_name}codec=plain{format=%{message}}}#stdout{codec=rubydebug}#warningoutput,ps:makesureattheendofoutputiftag_errorin[tags]{exec{command=echo'%{@timestamp}%{source}:%{message}'|mail-s$