ELK日志管理深度实战

Logstash收集Rsyslog/syslog日志

syslog默认是通过514端口去发送日志，可以使用logstash安装对应的插件，监听514端口来收集日志。如果只是监听系统日志可以不用安装logstash的agent,只需要监听对应的514端口，收集系统数据即可。

logstash在INPUT插件中提供了syslog的模块，可以用于不安装本地日志收集agent的设备（如硬件防火墙等），当然支持此类协议的普通服务器也适用。

注意：此INPUT插件会同时监听TCP和UDP端口。

可以在安装logstash服务器node2(172.16.10.21)上编写logstash配置文件,指定端口（默认为514）同时指定写入的ES服务器：

<code>[root@node2 ~]</code><code># cat /etc/logstash/conf.d/syslog.conf </code>

<code>input {</code>

<code> </code><code>syslog{</code>

<code> </code><code>type</code> <code>=> </code><code>"system-syslog"</code>

<code>output {</code>

<code> </code><code>elasticsearch {</code>

<code> </code><code>hosts => [</code><code>"172.16.10.21:9200"</code><code>] </code><code># 指定写入的ES服务器</code>

<code> </code><code>index => </code><code>"system-syslog-%{+YYYY.MM}"</code>

在node1(172.16.10.20)使syslog自动发送日志信息到logstash，需要修改/etc/rsyslog.conf （CentOS7）的配置文件：

在最后添加一行，指定远程logstash服务器地址和端口：

重启rsyslog服务：

<code>systemctl restart rsyslog</code>

在logstash服务器上指定配置文件启动：

<code>/opt/logstash/bin/logstash</code> <code>-f </code><code>/etc/logstash/conf</code><code>.d</code><code>/syslog</code><code>.conf</code>

在node1使用logger 命令可以生成系统日志，使Elasticsearch上产生索引：

<code>logger trying</code>

登录Elasticsearch上就可查询到syslog的日志了，可以在kibana上添加此项。

TCP、UDP日志收集

在有时要对一些日志进行补充或者发送一些文件内容到es上时，可以通过使用TCP模块的方式。

编辑tcp.conf的logstash文件：

<code>[root@node2 ~]</code><code># cat /etc/logstash/conf.d/tcp.conf </code>

<code> </code><code>mode => </code><code>"server"</code>

<code>output{</code>

<code> </code><code>stdout {</code>

<code> </code><code>codec => rubydebug</code>

启动服务，在其他任何主机上可以通过nc 命令或者伪设备输出的方式，将需要发送的内容发送给node2:

<code>[root@node1 ~]</code><code># echo "trying"|nc 172.16.10.21 6666</code>

<code>[root@node1 ~]</code><code># echo "stuff" > /dev/tcp/172.16.10.21/6666</code>

在node2上，显示接受的信息：

<code>[root@node2 ~]</code><code># /opt/logstash/bin/logstash -f /etc/logstash/conf.d/tcp.conf </code>

<code>Settings: Default pipeline workers: 1</code>

<code>Pipeline main started</code>

<code> </code><code>"message"</code> <code>=> </code><code>"trying"</code><code>,</code>

<code> </code><code>"@version"</code> <code>=> </code><code>"1"</code><code>,</code>

<code> </code><code>"@timestamp"</code> <code>=> </code><code>"2017-01-03T08:12:10.630Z"</code><code>,</code>

<code> </code><code>"message"</code> <code>=> </code><code>"# Generated by NetworkManager"</code><code>,</code>

<code> </code><code>"@timestamp"</code> <code>=> </code><code>"2017-01-03T08:13:11.081Z"</code><code>,</code>

<code> </code><code>"message"</code> <code>=> </code><code>"nameserver 114.114.114.114"</code><code>,</code>

<code> </code><code>"message"</code> <code>=> </code><code>"stuff"</code><code>,</code>

<code> </code><code>"@timestamp"</code> <code>=> </code><code>"2017-01-03T08:15:14.224Z"</code><code>,</code>

使用Filter模块对Apache日志进行格式化

在logstash中可以对一些输入的格式进行格式化，如记录Apache日志时，由于自身不具有像nginx提供的JSON格式化的输出，所以在进行记录时，需要使用filter插件的grok来进行正则匹配。

在logstash中，已经默认提供了方便使用的正则表示方式,在/opt/logstash/vendor/bundle/jruby/1.9/gems/logstash-patterns-core-2.0.5/patterns保存了相关的配置。

编写logstash的配置文件：

<code>[root@node2 patterns]</code><code># cat /etc/logstash/conf.d/grok.conf </code>

<code>input{</code>

<code> </code><code>stdin {}</code>

<code>filter { </code>

<code> </code><code>match => { </code><code>"message"</code> <code>=> </code><code>"%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}"</code> <code>}</code>

<code> </code><code>stdout{</code>

启动服务，输入测试示例，返回格式化的输出：

<code>[root@node2 ~]</code><code># /opt/logstash/bin/logstash -f /etc/logstash/conf.d/grok.conf </code>

<code>55.3.244.1 GET </code><code>/index</code><code>.html 15824 0.043 </code><code>#输入的示例</code>

<code> </code><code>"message"</code> <code>=> </code><code>"55.3.244.1 GET /index.html 15824 0.043"</code><code>,</code>

<code> </code><code>"@timestamp"</code> <code>=> </code><code>"2017-01-03T08:59:44.264Z"</code><code>,</code>

<code> </code><code>"client"</code> <code>=> </code><code>"55.3.244.1"</code><code>,</code>

<code> </code><code>"method"</code> <code>=> </code><code>"GET"</code><code>,</code>

<code> </code><code>"request"</code> <code>=> </code><code>"/index.html"</code><code>,</code>

<code> </code><code>"bytes"</code> <code>=> </code><code>"15824"</code><code>,</code>

<code> </code><code>"duration"</code> <code>=> </code><code>"0.043"</code>

使用grok收集Apache日志到Elasticsearch.

在默认的Apache日志格式下，logstash提供了一个已经编写好的日志格式正则filter:

<code>[root@node2 ~]</code><code># cat /opt/logstash/vendor/bundle/jruby/1.9/gems/logstash-patterns-core-2.0.5/patterns/grok-patterns \</code>

<code>|</code><code>grep</code> <code>"COMBINEDAPACHELOG"</code>

<code>COMBINEDAPACHELOG %{COMMONAPACHELOG} %{QS:referrer} %{QS:agent}</code>

编写logstash配置文件,引用此参数COMBINEDAPACHELOG：

<code>[root@node2 ~]</code><code># cat /etc/logstash/conf.d/grok.conf </code>

<code> </code><code>path => </code><code>"/var/log/httpd/access_log"</code>

<code> </code><code>start_position => </code><code>"beginning"</code>

<code> </code><code>match => { </code><code>"message"</code> <code>=> </code><code>"%{COMBINEDAPACHELOG}"</code> <code>} </code><code>#调用此参数</code>

<code> </code><code>hosts => [</code><code>"172.16.10.20:9200"</code><code>]</code>

<code> </code><code>index => </code><code>"apache-access-log-%{+YYYY.MM.dd}"</code>

<code># stdout{ codec => rubydebug }</code>

这样，默认的日志文件就可以被格式化输出到Elasticsearch上了。

使用grok会有一些问题，比如会非常影响性能，而且很不灵活，除非对ruby掌控很好。

在实际的生产环境中，为了实现松耦合和性能问题，一般将日志写入redis,再使用python对其进行格式化处理。

使用消息队列扩展ELK

数据 >> lostash >> MQ/Redis >> logstash >> ES

在实现低耦合和高可靠性的ELK部署架构中，使用消息队列的方式来作为一个中间转存的数据交换中心，将数据文件需要写入到消息队列，而对消息队列中数据的处理可有其它任意的应用去实现，整个过程写入和输出系统没有任何依赖关系。

这里使用最简单的redis来实现这一功能。

在node2上安装redis,将收集的数据发送到redis中，不做任何处理，这里需要先修改redis的配置文件：

<code># vim /etc/redis.conf </code>

<code>daemonize </code><code>yes</code> <code># 使用台运行模式</code>

<code>bind 172.16.10.21 </code><code># 绑定监听的IP</code>

通过使用logstash将数据写入redis,创建logstash的配置文件，用于将数据写入redis:

<code>[root@node2 ~]</code><code># cat /etc/logstash/conf.d/redis.conf </code>

<code> </code><code>stdin {</code>

<code> </code><code>redis {</code>

启动redis,指定配置文件启动logstash:

<code>systemctl start redis</code>

<code>[root@node2 ~]</code><code># /opt/logstash/bin/logstash -f /etc/logstash/conf.d/redis.conf </code>

在redis上可以查看到输入的数据：

<code>[root@node2 ~]</code><code># redis-cli -h 172.16.10.21 -p 6379</code>

<code># Keyspace</code>

<code>db6:keys=2,expires=0,avg_ttl=0</code>

<code>172.16.10.21:6379> </code><code>select</code> <code>6</code>

<code>1) </code><code>"apache-accesslog"</code>

<code>172.16.10.21:6379[6]> lindex demo -1</code>

<code>"{\"message\":\"try\",\"@version\":\"1\",\"@timestamp\":\"2017-01-04T03:29:28.337Z\",\"host\":\"node2\"}"</code>

使用另一个logstash将数据从redis中取出：

在node1上配置logstash的配置文件，指定文件启动，此时会自动将redis上的数据读出：

<code>[root@node1 ~]</code><code># cat /etc/logstash/conf.d/redis-get.conf </code>

<code> </code><code>redis {</code>

<code>filter {}</code>

<code>[root@node1 conf.d]</code><code># /opt/logstash/bin/logstash -f /etc/logstash/conf.d/redis-get.conf </code>

<code> </code><code>"message"</code> <code>=> </code><code>"try"</code><code>,</code>

<code> </code><code>"@timestamp"</code> <code>=> </code><code>"2017-01-03T12:10:20.635Z"</code><code>,</code>

上面使用的是手动输入的方式，在实际收集日志时，可以在输入时指定文件，输出时指定到Elasticsearch。

使用Redis收集Apache访问日志

在上面的示例中，我们使用了直接通过grok来收集Apache的日志，但是在实际的生产环境，由于这种方案会使性能下降，而且在整个架构上过度依赖于logstash,无法做到解耦，并不是一个好的解决方案。这里引入消息队列，作为一个解耦，上日志的存取分开，方便后期扩展，提高了稳定性。

在node2上配置logstash 配置文件，指定Apache的access文件，并启动：

<code>[root@node2 ~]</code><code># cat /etc/logstash/conf.d/apache-redis.conf </code>

<code> </code><code>key => </code><code>"apache-log"</code>

此处直接完整输入内容到redis,不做任何处理。

在node1上从redis中取出日志，并通过grok进行日志格式化处理，之后输出到Elasticsearch中。

node1上logstash的配置：

<code>filter {</code>

<code> </code><code>match => { </code><code>"message"</code> <code>=> </code><code>"%{COMBINEDAPACHELOG}"</code> <code>}</code>

<code> </code><code>elasticsearch{</code>

<code> </code><code>index => </code><code>"apache-log-%{+YYYY.MM.dd}"</code>

这样，启动node1和node2上的logstash,在指定的Elasticsearch上就可以看到被格式化后的日志输出了。

ELK项目生产规划

在实际的生产环境中要对日志进行收集，首先需要对日志分类：

访问日志： apache，nginx, tomcat 等web访问日志通过file插件收集，apache通过filter插件处理格式

错误日志：error log,java 日志等可以直接收取，对于一条多行的日志（java异常）使用多行处理插件

系统日志： /var/log syslog 等直接使用 syslog插件收取

网络日志：防火墙，交换机，路由器等使用 syslog日志

运行日志：应用程序定义使用file 插件，最好为json格式

标准化：统一存放路径，命名规则，格式（json）,日志切割（crontab）. 需要保存的日志先在本地读写，提高性能，通过rsync推送到指定存储，防止挂载故障而使服务终止。删除本地日志（确认保留时间）

工具化：可以使用logstash进行日志收集方案。

如果使用redis作为消息队列，那么要对所有的list key的长度进行监控，防止溢出和数据丢失

llen key_name

根据实际情况，例如超过 10万就报警。

如将之前的所有配置合并到一处，在使用redis作为存储时，可以这样编辑logstash配置文件：

从redis服务器接受日志,处理后存入Elasticsearch：

<code> </code><code>syslog {</code>

<code> </code><code>redis {</code>

<code> </code><code>type</code> <code>=> </code><code>"apache-accesslog"</code>

<code> </code><code>key => </code><code>"apache-accesslog"</code>

<code> </code><code>redis {</code>

<code> </code><code>if</code> <code>[</code><code>type</code><code>] == </code><code>"apache-accesslog"</code> <code>{</code>

<code> </code><code>match => { </code><code>"message"</code> <code>=> </code><code>"%{COMBINEDAPACHELOG}"</code> <code>}</code>

<code> </code><code>hosts => [</code><code>"192.168.56.11:9200"</code><code>]</code>

<code> </code><code>index => </code><code>"apache-accesslog-%{+YYYY.MM.dd}"</code>

<code> </code><code>elasticsearch {</code>

<code> </code><code>hosts => [</code><code>"192.168.56.11:9200"</code><code>]</code>

<code> </code><code>index => </code><code>"es-log-%{+YYYY.MM}"</code>

<code> </code><code>if</code> <code>[</code><code>type</code><code>] == </code><code>"system-syslog"</code> <code>{</code>

<code> </code><code>index => </code><code>"system-syslog-%{+YYYY.MM}"</code>

从服务器接收日志，将日志存入redis/MQ:

<code> </code><code>path => </code><code>"/var/log/elasticsearch/elasticsearch.log"</code>

<code> </code><code>codec => multiline{</code>

<code> </code><code>pattern => </code><code>"^\["</code>

<code> </code><code>negate => </code><code>true</code>

<code> </code><code>what => </code><code>"previous"</code>

<code> </code><code>key => </code><code>"apache-accesslog"</code>

这里使用了if做为了条件的判断，来对不同的日志进行选择处理，在一些系统日志如syslog等，可以单独进行处理。

提示：在使用系统自带的脚本进行启动时，会出现无法收集日志的情况，脚本中默认的是使用logstash用户，在收集一些日志时可能存在没有权限的情况。可以修改脚本中的默认用户，或者给logstash添加其他用户组的权限。

本文转自酥心糖 51CTO博客，原文链接:http://blog.51cto.com/tryingstuff/1888978

ELK日志管理深度实战

继续阅读

httpd dead but subsys locked;No space left on device:Couldn't create accept loc

apache (httpd)不支持中文路径问题先卸载yum安装的httpd再用源码安装，重装httpd再安装支持中文的插件遇到问题

搭建httpd服务

windows下配置Apache的vhost初次接触，强烈欢迎拍砖，指出错误

Apache与PHP环境下配置本地虚拟主机

Linux 7 中配置Apache服务，及禁止ip访问，删除apache广告页面。

Apache配置文件中的deny和allow的使用

Apache 配置默认编码

服务器配置——Apache

Apache静态文件访问配置（书封服务器）

apache httpd 配置

Ubuntu16.04安装Apache+MySQL+PHP1. 安装Apache2. 安装MySQL3. 安装PHP4. 安装phpMyAdmin

Apache配置SSLApache配置SSL

Windows下配置Apache的SSL服务

Apache2.4.x 配置文件详解Apache配置需要了解如下：开始讲解：

配置apache支持PHP（win7）