首页 文章 精选 留言 我的

ELK日志管理深度实战

2017-11-11 638

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服务器：

1
2
3
4
5
6
7
8
9
10
11
12
13
[root@node2 ~] # cat /etc/logstash/conf.d/syslog.conf 
input {
      syslog{
      type  =>  "system-syslog"
      port => 514
}
}
output {
     elasticsearch {
         hosts => [ "172.16.10.21:9200" ]          # 指定写入的ES服务器
         index =>  "system-syslog-%{+YYYY.MM}"
     }
}

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

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

1
*.* @@172.16.10.21:514

重启rsyslog服务：

1
systemctl restart rsyslog

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

1
/opt/logstash/bin/logstash  -f  /etc/logstash/conf .d /syslog .conf

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

1
logger trying

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


TCP、UDP日志收集

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

编辑tcp.conf的logstash文件：

1
2
3
4
5
6
7
8
9
10
11
12
13
[root@node2 ~] # cat /etc/logstash/conf.d/tcp.conf 
input {
     tcp {
         type  =>  "tcp"
         port =>  "6666"
         mode =>  "server"
    }
}
output{
     stdout {
         codec => rubydebug
}
}

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

1
2
3
[root@node1 ~] # echo "trying"|nc 172.16.10.21 6666
[root@node1 ~] # nc 172.16.10.21 6666 < /etc/resolv.conf 
[root@node1 ~] # echo "stuff" > /dev/tcp/172.16.10.21/6666

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
[root@node2 ~] # /opt/logstash/bin/logstash -f /etc/logstash/conf.d/tcp.conf 
Settings: Default pipeline workers: 1
Pipeline main started
{
        "message"  =>  "trying" ,
       "@version"  =>  "1" ,
     "@timestamp"  =>  "2017-01-03T08:12:10.630Z" ,
           "host"  =>  "172.16.10.20" ,
           "port"  => 57054,
           "type"  =>  "tcp"
}
{
        "message"  =>  "# Generated by NetworkManager" ,
       "@version"  =>  "1" ,
     "@timestamp"  =>  "2017-01-03T08:13:11.081Z" ,
           "host"  =>  "172.16.10.20" ,
           "port"  => 57248,
           "type"  =>  "tcp"
}
{
        "message"  =>  "nameserver 114.114.114.114" ,
       "@version"  =>  "1" ,
     "@timestamp"  =>  "2017-01-03T08:13:11.081Z" ,
           "host"  =>  "172.16.10.20" ,
           "port"  => 57248,
           "type"  =>  "tcp"
}
{
        "message"  =>  "stuff" ,
       "@version"  =>  "1" ,
     "@timestamp"  =>  "2017-01-03T08:15:14.224Z" ,
           "host"  =>  "172.16.10.20" ,
           "port"  => 57634,
           "type"  =>  "tcp"
}


使用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的配置文件：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
[root@node2 patterns] # cat /etc/logstash/conf.d/grok.conf 
input{
     stdin {}
}
filter { 
      grok {
     match => {  "message"  =>  "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}"  }
   }
}
output {
     stdout{
         codec => rubydebug
}
}

官方文档提供了示例，可以直接使用https://www.elastic.co/guide/en/logstash/current/plugins-filters-grok.html 

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
[root@node2 ~] # /opt/logstash/bin/logstash -f /etc/logstash/conf.d/grok.conf 
Settings: Default pipeline workers: 1
Pipeline main started
55.3.244.1 GET  /index .html 15824 0.043   #输入的示例
{
        "message"  =>  "55.3.244.1 GET /index.html 15824 0.043" ,
       "@version"  =>  "1" ,
     "@timestamp"  =>  "2017-01-03T08:59:44.264Z" ,
           "host"  =>  "node2" ,
         "client"  =>  "55.3.244.1" ,
         "method"  =>  "GET" ,
        "request"  =>  "/index.html" ,
          "bytes"  =>  "15824" ,
       "duration"  =>  "0.043"
}

使用grok收集Apache日志到Elasticsearch.


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

1
2
3
4
[root@node2 ~] # cat /opt/logstash/vendor/bundle/jruby/1.9/gems/logstash-patterns-core-2.0.5/patterns/grok-patterns \
| grep  "COMBINEDAPACHELOG"
 
COMBINEDAPACHELOG %{COMMONAPACHELOG} %{QS:referrer} %{QS:agent}


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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
[root@node2 ~] # cat /etc/logstash/conf.d/grok.conf                              
input{
     file  {
         path =>  "/var/log/httpd/access_log"
         start_position =>  "beginning"
}
}
filter { 
      grok {
     match => {  "message"  =>  "%{COMBINEDAPACHELOG}"  }   #调用此参数
   }
}
output {
     elasticsearch {
         hosts => [ "172.16.10.20:9200" ]
         index =>  "apache-access-log-%{+YYYY.MM.dd}"
}
#   stdout{ codec => rubydebug  }
}

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


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

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


使用消息队列扩展ELK


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

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

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

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

1
2
3
# vim /etc/redis.conf 
daemonize  yes           # 使用台运行模式
bind 172.16.10.21       # 绑定监听的IP

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
[root@node2 ~] # cat /etc/logstash/conf.d/redis.conf 
input{
     stdin {
}
}
output {
         redis {
           host =>  "172.16.10.21"
           port =>  "6379"
           db =>  "6"
           data_type =>  "list"
           key =>  "demo"
}
}

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

1
systemctl start redis
1
2
3
4
[root@node2 ~] # /opt/logstash/bin/logstash -f /etc/logstash/conf.d/redis.conf 
Settings: Default pipeline workers: 1
Pipeline main started
try

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

1
2
3
4
5
6
7
8
9
10
11
[root@node2 ~] # redis-cli  -h 172.16.10.21 -p 6379
172.16.10.21:6379> info
# Keyspace
db6:keys=2,expires=0,avg_ttl=0
172.16.10.21:6379>  select  6
OK
172.16.10.21:6379[6]> keys *
1)  "apache-accesslog"
2)  "demo"
172.16.10.21:6379[6]> lindex demo -1
"{\"message\":\"try\",\"@version\":\"1\",\"@timestamp\":\"2017-01-04T03:29:28.337Z\",\"host\":\"node2\"}"

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

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
[root@node1 ~] # cat /etc/logstash/conf.d/redis-get.conf 
input {
        redis {
           host =>  "172.16.10.21"
           port =>  "6379"
           db =>  "6"
           data_type =>  "list"
           key =>  "demo"
}
}
filter {}
output {
     stdout{
         codec => rubydebug
}
}
1
2
3
4
5
6
7
8
9
[root@node1 conf.d] # /opt/logstash/bin/logstash -f /etc/logstash/conf.d/redis-get.conf 
Settings: Default pipeline workers: 1
Pipeline main started
{
        "message"  =>  "try" ,
       "@version"  =>  "1" ,
     "@timestamp"  =>  "2017-01-03T12:10:20.635Z" ,
           "host"  =>  "node2"
}

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


使用Redis收集Apache访问日志

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

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
[root@node2 ~] # cat /etc/logstash/conf.d/apache-redis.conf 
input{
     file  {
         path =>  "/var/log/httpd/access_log"
         start_position =>  "beginning"
}
}
output {
         redis {
           host =>  "172.16.10.21"
           port =>  "6379"
           db =>  "6"
           data_type =>  "list"
           key =>  "apache-log"
}
}

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

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

node1上logstash的配置：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
[root@node1 ~] # cat /etc/logstash/conf.d/redis-get.conf 
input {
        redis {
           host =>  "172.16.10.21"
           port =>  "6379"
           db =>  "6"
           data_type =>  "list"
           key =>  "apache-log"
}
}
filter {
     grok {
     match => {  "message"  =>  "%{COMBINEDAPACHELOG}"  }
   }
}
output {
     elasticsearch{
         hosts => [ "172.16.10.20:9200" ]
         index =>  "apache-log-%{+YYYY.MM.dd}"
}
}

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

和redis类似，可以使用Kafka作为存储数据的中间节点：https://www.unixhot.com/article/61 



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：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
input {
    syslog {
      type  =>  "system-syslog"
      port => 514
    }
    redis {
         type  =>  "apache-accesslog"
         host =>  "192.168.56.12"
         port =>  "6379"
         db =>  "6"
         data_type =>  "list"
         key =>  "apache-accesslog"
     }
     redis {
         type  =>  "es-log"
         host =>  "192.168.56.12"
         port =>  "6379"
         db =>  "6"
         data_type =>  "list"
         key =>  "es-log"
     }
}
filter {
     if  [ type ] ==  "apache-accesslog"  {
     grok {
         match => {  "message"  =>  "%{COMBINEDAPACHELOG}"  }
     }
     }
}
output {
     if  [ type ] ==  "apache-accesslog"  {
     elasticsearch {
        hosts => [ "192.168.56.11:9200" ]
         index =>  "apache-accesslog-%{+YYYY.MM.dd}"
     }
     }
     if  [ type ] ==  "es-log"  {
         elasticsearch {
                 hosts => [ "192.168.56.11:9200" ]
                 index =>  "es-log-%{+YYYY.MM}"
         }
     }
     if  [ type ] ==  "system-syslog"  {
         elasticsearch {
                 hosts => [ "192.168.56.11:9200" ]
                 index =>  "system-syslog-%{+YYYY.MM}"
         }
     }
}


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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
input {
     file  {
         path =>  "/var/log/httpd/access_log"
         start_position =>  "beginning"
         type  =>  "apache-accesslog"
     }
     file  {
         path =>  "/var/log/elasticsearch/elasticsearch.log"
         type  =>  "es-log"
         start_position =>  "beginning"
         codec => multiline{
           pattern =>  "^\["
           negate =>  true
           what =>  "previous"
         }
     }
}
output {
     if  [ type ] ==  "apache-accesslog"  {
         redis {
                 host =>  "192.168.56.12"
                 port =>  "6379"
                 db =>  "6"
                 data_type =>  "list"
                 key =>  "apache-accesslog"
         }
     }
     if  [ type ] ==  "es-log"  {
     redis {
         host =>  "192.168.56.12"
         port =>  "6379"
         db =>  "6"
         data_type =>  "list"
         key =>  "es-log"
     }
     }
}

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


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



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


上一篇 下一篇
优秀的个人博客,低调大师

微信关注我们

原文链接：https://yq.aliyun.com/articles/491255

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

Spark 算子

==>RDD是什么? ---> RDD(Resilient Distributed Dataset)弹性分布式数据集，是Spark中最基本的数据抽象，它代表一个不可变，可分区，里面的元素可并行计算的集合 --->特点： ----自动容错 ----位置感知性高度 ----可伸缩性 ----允许用户在执行多个查询时显示的将工作集缓存在内存中，后续的查询能够重用工作集，极大的提升了查询速度 --->RDD的属性 ----A list of partitions 一个组分片，即数据集的基本组成单位 对于RDD来说，每个分片都会被一个计算任务处理，并决定并行计算的粒度，用户可以在创建RDD时指定 RDD的分片个数，如果没有指定，那么就会采用默认值，默认值就是程序所分配到的CPU Core的数目 ---- A function for computing each split 一个计算每个分区的函数 Spark中RDD的计算是以分片为单位的，每个RDD都会实现 compute函数以达到这个目的，compute函数会对迭代器进行复合，不需要保存每次...
2017-11-12
622
上一篇

使用 Scala 写WordContext程序

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 package mydemo import org.apache.spark.{SparkConf,SparkContext} object MyWordContextDemo{ def main(args : Array[String]) : Unit = { //创建一个Config val conf = new SparkConf().setAppName( "MyWordContext" ) //创建SparkContext对象 val sc = new SparkContext(conf) //使用sc对象执行相应的算子 sc.textFile(args( 0 )) //从第一个参数获取文件路径并打开 .flatMap( _ .split( "" )) //先map再压平 .map(( _ , 1 )) //将单词和1构成元组 .reduceByKey( _ + _ ) //根据Key进行Reduce，并将value累...
2017-11-12
700
下一篇

相关文章

发表评论

资源下载

更多资源
Mario

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长，特征是大鼻子、头戴帽子、身穿背带裤，还留着胡子。与他的双胞胎兄弟路易基一起，长年担任任天堂的招牌角色。

时间: 2025-12-12 大小: 211.28KB 下载: 54次
腾讯云软件源

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题，腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构，目前腾讯云软件源站支持公网访问和内网访问。

时间: 2026-03-01 大小: 1MB 下载: 6次
Nacos

Nacos

Nacos /nɑ:kəʊs/ 是 Dynamic Naming and Configuration Service 的首字母简称，一个易于构建 AI Agent 应用的动态服务发现、配置管理和AI智能体管理平台。Nacos 致力于帮助您发现、配置和管理微服务及AI智能体应用。Nacos 提供了一组简单易用的特性集，帮助您快速实现动态服务发现、服务配置、服务元数据、流量管理。Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。

时间: 2025-12-21 大小: 189MB 下载: 2次
Spring

Spring

Spring框架（Spring Framework）是由Rod Johnson于2002年提出的开源Java企业级应用框架，旨在通过使用JavaBean替代传统EJB实现方式降低企业级编程开发的复杂性。该框架基于简单性、可测试性和松耦合性设计理念，提供核心容器、应用上下文、数据访问集成等模块，支持整合Hibernate、Struts等第三方框架，其适用范围不仅限于服务器端开发，绝大多数Java应用均可从中受益。

时间: 2025-12-17 大小: 5MB 下载: 1次
手机查看

扫码在手机上查看文章

扫描二维码，手机阅读更方便

热门标签
Kubernetes DeepSeek Docker Jdk25 Rocky HarmonyOS6 Redis OpenAi SpringCloud Gemini3 Service Mesh SpringBoot4 Nacos3

欢迎您来访！

登录后，您将获得更多功能！
热门文章
联系我们

有任何问题或合作意向欢迎联系我们

Email: 99873273@qq.com

QQ: 99873273

用户登录
用户注册