ELK日志收集

[db:作者]

搭建ELK
ELK是由elasticsearch、logstash、kibana三个开源软件组成的一个组合体，ELK是elastic公司公司研发的一套完整的日志收集、分析和展示的企业级解决方案，在这三个软件当中，每个软件用于完成不同的功能，官方域名为elastic.io，ELK stack的主要优点：

处理方式灵活：elasticsearch是实时全文索引，具有强大的搜索功能配置相当简单：elasticsearch的API全部使用JSON接口，logstash使用模块配置，kibana的配置文件部分更简单检索性能高效：基于优秀的设计，虽然每次查询都是实时，但是也可以达到百亿数据的查询秒级响应。集群线性扩展：elasticsearch和logstash都可以灵活线性扩展前端操作绚丽：kibana的前端设计比较绚丽，而且操作简单
Elasticsearch
elasticsearch是一个高度可扩展的开源全文搜索和分析引擎，它可实现数据的实时全文搜索、支持分布式可实现高可用、提供API接口，可以处理大规模日志数据，比如nginx、tomcat、系统日志等功能。
- j7 p: @2 s" a! t: D
elasticsearch的特点：
" p6 X. m# x7 {
+ {. k* n6 e- d" H6 p6 e4 j  N实时收索、实时分析分布式架构、实时文件存储文档导向，所有对象都是文档高可用，易扩展，支持集群，分片与复制接口友好，支持json
部署elasticsearch
GitHub - elastic/elasticsearch: Free and Open, Distributed, RESTful Search Engine，基于java开发
9 W: c% E/ o) z  U0 ?1 ?1 A6 i
centos系统关闭服务器的防火墙和selinux，ubuntu关闭防火墙，保持各服务器时间同步
) Q. m* D4 a/ v' ^- i3 t; t5 A
" Q5 p# n" n/ a$ V服务器1:172.20.22.24
% F( l" C  N7 U& Z
服务器2:172.20.22.27

9 t. _* s1 G- `服务器3:172.20.22.28

###ubuntu
# apt install -y ntpdate
- W+ Z9 h2 p" L/ }. R# rm -f /etc/localtime
+ `: m! o+ e4 m/ h# c( C, _# ln -s /usr/share/zoneinfo/Asia/Shanghai /etc/localtime
# hwclock --systohc
# ntpdate -u ntp1.aliyun.com
3 B: S, t# U$ h* A###设置内核参数
# vim /etc/security/limits.conf
*                soft        nofile                500000
( @7 j, N9 r% K*                hard        nofile                500000
# vim /etc/security/limits.d/20-nproc.conf
1 N2 u8 Z% E* z2 N*          soft    nproc     4096
elasticsearch soft    nproc     unlimited
' F+ i( e1 D: Froot       soft    nproc     unlimited
###安装jdk
# apt install -y openjdk-8-jdk
; _; U3 x& _  q0 J! g/ h6 R" ^1 }
###每个节点都安装
  T7 J) u: P# d% h3 c# ls -lrt elasticsearch-7.12.1-amd64.deb
# dpkg -i elasticsearch-7.12.1-amd64.deb
###节点1配置文件
! B! m* j- d( c- s# grep '^[^#]' /etc/elasticsearch/elasticsearch.yml
cluster.name: m63-elastic        #集群名称
" @$ j- _9 i2 M5 Y+ l& z  Bnode.name: node1                 #当前节点在集群内的节点名称
path.data: /data/elasticsearch   #数据保存目录
path.logs: /data/elasticsearch   #日志保存目录
/ {. ]/ p3 c) }  lbootstrap.memory_lock: true      #服务启动的时候锁定足够的内存，防止数据写入swap
) ~! Y3 P7 ^4 w# V* T8 Fnetwork.host: 172.20.22.24       #监听IP
http.port: 9200                  #监听端口
' F" t6 ~3 @; ?& q$ c###集群中node节点发现列表
discovery.seed_hosts: ["172.20.22.24", "172.20.22.27","172.20.22.28"]
) G! r8 W3 W5 n' M7 {###集群初始化哪些节点可以被选举为master
7 o: l2 C% _. I1 W/ ~cluster.initial_master_nodes: ["172.20.22.24", "172.20.22.27","172.20.22.28"]
action.destructive_requires_name: true
# T  v$ J1 }, C9 j# mkdir /data/elasticsearch -p
1 U0 g3 q( A! I+ D( R! q, ]# chown -R elasticsearch. /data/elasticsearch
# systemctl start elasticsearch.service
###节点2
# grep '^[^#]' /etc/elasticsearch/elasticsearch.yml
  ^2 M# J: U' P6 icluster.name: m63-elastic
node.name: node2
2 G7 D( i" |$ M2 cpath.data: /data/elasticsearch
path.logs: /data/elasticsearch
network.host: 172.20.22.27
http.port: 9200
discovery.seed_hosts: ["172.20.22.24", "172.20.22.27","172.20.22.28"]
cluster.initial_master_nodes: ["172.20.22.24", "172.20.22.27","172.20.22.28"]
action.destructive_requires_name: true
# mkdir /data/elasticsearch -p
# chown -R elasticsearch. /data/elasticsearch
# systemctl start elasticsearch.service
###节点3
) E' q& A+ f$ ~& G. F' J# grep '^[^#]' /etc/elasticsearch/elasticsearch.yml
8 f1 Y" q* [0 W5 acluster.name: m63-elastic
0 }( t8 v! s2 }node.name: node3
; f& x% {4 n! ]( b& ^path.data: /data/elasticsearch
; Q" K* t$ c: @path.logs: /data/elasticsearch
network.host: 172.20.22.28
http.port: 9200
! X: `+ c% c2 m4 i' [! s5 Pdiscovery.seed_hosts: ["172.20.22.24", "172.20.22.27","172.20.22.28"]
% M: i0 ]. J: qcluster.initial_master_nodes: ["172.20.22.24", "172.20.22.27","172.20.22.28"]
action.destructive_requires_name: true
5 P5 p$ X' N! }+ C3 f/ H  Q# mkdir /data/elasticsearch -p
# chown -R elasticsearch. /data/elasticsearch
, i; V( D( F; B6 u8 q  l0 i& y# systemctl start elasticsearch.service
% k. e: `+ ^7 H. F浏览器访问验证
" c; v9 e( i' f( o' }2 G" z& Uhttp://$IP:9200


2 e, B- e& o6 ?0 E% A0 X9 ~
, U+ p3 {$ U+ l) p8 c7 Z6 Q* eLogstash
5 f; V0 D3 b$ a- B5 H/ g# h/ wLogstash是一个具有实时传输能力的数据收集引擎，其可以通过插件实现日志收集和转发，支持日志过滤，支持普通log、自定义json格式的日志解析，最终把经过处理的日志发送给elasticsearch。
$ r6 ~+ [. f9 h5 x5 u5 ?
部署Logstash
Logstash是一个开源的数据收集引擎，可以水平伸缩，而且logstash是整个ELK当中用于最多插件的一个组件，其可以接收来自不同来源的数据并统一输出到指定的且可以是多个不同目的地

https://github.com/elastic/logstash #GitHub
+ _/ F: d  f, b- k
0 y# Q; r2 L8 B0 r# b0 jElastic Stack and Product Documentation | Elastic
, X9 h2 z" f  }. R( v' [6 V
环境准备：关闭防火墙和selinux，并且安装java环境

9 C  ?/ w6 b. _" D% @& [# apt install -y openjdk-8-jdk
) \5 Q: z1 R$ H! F( ?+ h# ls -lrt logstash-7.12.1-amd64.deb
7 \2 I- _/ V! R" ~) H# dpkg -i logstash-7.12.1-amd64.deb
9 ~; R5 J3 N  l4 h0 I###启动测试
# /usr/share/logstash/bin/logstash -e  'input { stdin {} } output { stdout {}}'   ##标准输入和标准输出
" B1 Z, w  q' w" @* lhello world!~
  N5 t' ?" F5 P( z{
3 f+ d* l) k  B8 _8 o      "@version" => "1",
* H! B. _2 E) X0 a) [    "@timestamp" => 2022-04-13T06:16:32.212Z,
          "host" => "jenkins-slave",
       "message" => "hello world!~"
}
) y$ M! S  w! t###通过配置文件启动
) F5 C- _! p  N. |  R# cd /etc/logstash/conf.d/
% b- m6 ~  r' Q0 j# cat test.conf
input {
7 x% @! d5 }: k  stdin {}
& A/ b0 _& t  ^4 l! D1 d}
output {
$ U7 {5 z, f) [3 q5 R: d  stdout {}
}
6 x+ Q( z- P6 P( B9 O% V
' u0 s- r4 |/ n4 k. [###通过指定配置文件启动
# /usr/share/logstash/bin/logstash -f test.conf -t   ##检查配置文件语法
3 V" A8 z6 F* i( u# /usr/share/logstash/bin/logstash -f test.conf
0 Q" X& m* \# c6 V8 }# |0 K
; r4 U) v; J9 `' q# j3 I####输出到elasticsearch
1 j: I. n6 |; N. {! m- z# cat test.conf
; Q8 f( T  `' k& L3 y% v& Tinput {
  stdin {}
}
output {
  #stdout {}
  elasticsearch {
    hosts => ["172.20.22.24:9200"]
    index => "magedu-m63-test-%{+YYYY.MM.dd}"
  }
' K+ E9 K/ N1 D+ e6 q0 B}
7 e! j+ t# S# I- v) O# /usr/share/logstash/bin/logstash -f test.conf
version1
# ?! E& b% [$ q* a' T% v. b8 Oversion2
: m- p0 z4 U$ c7 n$ O- Nversion3
* t4 u; d5 T! a) G7 D. mtest1
6 [% L; o4 T4 |& Btest2
& T  `6 G: x# \) e3 t- `" l( ^7 gtest3
( Q0 D: C% Y. i! ]) H) d
" a9 x1 Y+ ?4 y+ e+ Q####elasticsearch服务器查看收集到的数据
# ls -lrt /data/elasticsearch/nodes/0/indices/
5 ^6 n! e4 @. ^; `. z- ^! ctotal 4
drwxr-xr-x 4 elasticsearch elasticsearch 4096 Apr 13 14:36 DyCv8w7mTleuAvlItAJlWA
kibana
- f. B! l6 ]+ T: K* H8 s  Okibana为elasticsearch提供一个查看数据的web界面，其主要是通过elasticsearch的API接口进行数据查找，并进行前端数据可视化的展现，另外还可以针对特定格式的数据生成相应的表格、柱状图、饼图等
* U* v7 N. `9 J6 ~! L2 A6 |
部署kibana
# ls -lrt kibana-7.12.1-amd64.deb
0 I* z, `, S5 n9 W. ?( {0 H# dpkg -i kibana-7.12.1-amd64.deb
# grep "^[^$|#]" /etc/kibana/kibana.yml
9 D6 S7 L. x" F8 j# {server.port: 5601
server.host: "172.20.22.24"
, Q3 i. E, j! J" a& b+ D. b, f& Delasticsearch.hosts: ["http://172.20.22.27:9200"]
+ E! b& v4 P/ Di18n.locale: "zh-CN"
# systemctl restart kibana
9 v: z& S+ L$ D, g浏览器访问http://172.20.22.24:5601
* q/ M( Q7 x/ ?# c
( w5 K8 E4 p; ~& kStack Management-->索引模式-->创建索引模式


选择时间字段
! f6 H& Y2 ?9 J6 ?/ [0 g, \/ z& ~/ O& f
8 [6 S5 [; @$ P/ m2 d/ y查看对应创建的索引日志信息



- }+ p, x% y' ^+ O7 k6 e. H- B收集tomcat日志
收集tomcat服务器的访问日志以及tomcat错误日志进行实时统计，在kibana页面进行搜索展现，每台tomcat服务器要安装logstash负责收集日志，然后将日志转发给elasticsearch进行分析，再通过kibana在前端展现

* z% x$ {2 G2 `- U) t, ^' e& H部署tomcat
####tomcat1,172.20.22.30
# apt install -y openjdk-8-jdk
) S) K# _& h  J$ z" O# ls -lrt apache-tomcat-8.5.77.tar.gz
-rw-r--r-- 1 root root 10559655 Apr 13 21:44 apache-tomcat-8.5.77.tar.gz
0 I. X* ?- e( ^" l% ~% l" b7 W# tar xf apache-tomcat-8.5.77.tar.gz -C /usr/local/src/
' b; R, M' p/ H& b# ln -s /usr/local/src/apache-tomcat-8.5.77 /usr/local/tomcat
  K4 z' P+ P- X4 ?: k) X- j  w$ Q9 Z4 L( v# cd /usr/local/tomcat
###修改tomcat日志格式为json
# vim conf/server.xml
....
) h# j& U' c- ~5 S9 `* M
, L0 q  w7 r' z5 K6 d: O. a; u....
# mkdir /usr/local/tomcat/webapps/myapp
# echo "web1 172.20.22.30" > /usr/local/tomcat/webapps/myapp/index.html
# ./bin/catalina.sh start

###访问测试
& t" q' \, }# C6 C+ x& c/ H! o; Q# curl http://172.20.22.30:8080/myapp/
###查看访问日志
# tail -f /usr/local/tomcat/logs/tomcat_access_log.2022-04-13.log
/ D  f1 y1 y( S$ U5 h  B7 a
####tomcat2,172.20.22.26
  I  M! M7 w2 d# apt install -y openjdk-8-jdk
# ls -lrt apache-tomcat-8.5.77.tar.gz
0 ~* E4 \! `$ W8 w0 n6 g1 K-rw-r--r-- 1 root root 10559655 Apr 13 21:44 apache-tomcat-8.5.77.tar.gz
& ]1 K; B9 ]  c# tar xf apache-tomcat-8.5.77.tar.gz -C /usr/local/src/
# ln -s /usr/local/src/apache-tomcat-8.5.77 /usr/local/tomcat
# cd /usr/local/tomcat
4 W' w' h  E1 ^/ ]###修改tomcat日志格式为json
% Y0 L1 d% h9 ]1 q5 n# vim conf/server.xml
....
# f! ]' K& K% @
7 ~! }' c7 W) J5 r# q3 D6 `# o....
# mkdir /usr/local/tomcat/webapps/myapp
: W2 Y8 m' Q! ~/ `# echo "web2 172.20.22.26" > /usr/local/tomcat/webapps/myapp/index.html
( h, V3 O* C5 a( R# ./bin/catalina.sh start

+ z4 C5 Z1 k. p  O  H- M7 d###访问测试
8 x) Y8 y, G1 O& }. Q* G8 E: a# curl http://172.20.22.26:8080/myapp/
9 O+ Z8 s+ D) _6 t" c$ h* ~###查看访问日志
: i8 q4 L, e, j# tail -f /usr/local/tomcat/logs/tomcat_access_log.2022-04-14.log
部署logstash
% j  R0 j1 ]+ E; m6 e' x: X- M7 l在tomcat服务器安装logstash收集tomcat和系统日志
6 [( Q9 p3 T6 B" z2 W( i8 T0 N  v
####tomcat1,172.20.22.30
# ls -lrt logstash-7.12.1-amd64.deb
  V9 o$ }# |8 \' ]# dpkg -i logstash-7.12.1-amd64.deb
# vim /etc/systemd/system/logstash.service
  h3 s6 p" a  f  O5 ^...
User=root
4 I; c$ Y8 Y9 E0 s( w# f4 LGroup=root
...
" E: {6 ?- d1 y( }# cd /etc/logstash/conf.d
# cat tomcat.conf
input {
* n( B: q( U1 A- l( Q+ v; c- \  file {
    path => "/usr/local/tomcat/logs/tomcat_access_log*.log"
& r( u: b! ?- B: K    type => "tomcat-log"
& u- f7 R. y. H# P% w    start_position => "beginning"
    stat_interval => "3"
  }
: q! _* S- w' w. _9 Z) J  file {
" m$ A& a' F/ w. E$ ]' \    path => "/var/log/syslog"
    type => "systemlog"
    start_position => "beginning"
    stat_interval => "3"
( W5 s$ y5 C! |* Q1 D% j  }
( Z8 |; z% \% [. s& X1 _/ V}
* L& g/ t; n& P! A3 |6 ooutput {
' Q& P- M. \. r/ A  if [type] == "tomcat-log" {
' {7 C, u4 i' i  elasticsearch {
: d+ m7 Q' }1 X% r& w3 n    hosts => ["172.20.22.24:9200","172.20.22.27:9200"]
    index => "elk-tomcat-%{+YYYY.MM.dd}"
) J* G  q* i$ l. s2 i" c7 Y8 k  }}
$ \$ U. L  z+ x8 T5 Z  Z/ @8 ]% l/ h  if [type] == "systemlog" {
  elasticsearch {
7 D$ `; J0 O" u0 p5 D1 ^; y    hosts => ["172.20.22.27:9200","172.20.22.27:9200"]
    index => "elk-syslog-%{+YYYY.MM.dd}"
- x4 ~* B: O2 ]  }}
) ]: {, [" s: u2 s}
* J9 b' B  z4 R5 @
# s* I% Z4 u* A! \) I# /usr/share/logstash/bin/logstash -f tomcat.conf -t
# systemctl daemon-reload
# systemctl start logstash.service
1 k* {* T' [9 v) s% F  a# scp tomcat.conf root@3172.20.22.26
. X, |& }' @$ o
####tomcat2,172.20.22.26
# ls -lrt logstash-7.12.1-amd64.deb
  M+ R/ X) v% Z5 Y# dpkg -i logstash-7.12.1-amd64.deb
# vim /etc/systemd/system/logstash.service
...
' q& \% a8 v4 Q6 R/ C' Y4 @User=root
Group=root
: \6 Q' @% _! C# A" O4 M...
. t3 D( b! W3 j( a; ^! c: M# systemctl daemon-reload
9 l. n& o: e0 f+ p0 s) f8 |* n# systemctl daemon-reload
( N5 d/ t9 X+ \) [" A, ~* e# systemctl start logstash.service
通过kibana展现

) I  r) k: t$ X% ~/ n
收集Java日志
使用codec的multiline插件实现多行匹配，这是一个可以将多行进行合并的插件，而且可以使用what指定将匹配到的行与前面的行合并还是和后面的行合并

6 I8 c1 ^$ e# Y7 x) \1 c- w$ FMultiline codec plugin | Logstash Reference [8.1] | Elastic
; L# L5 \( }2 K$ q
0 S! p% \9 G3 e# g# P添加logstash配置文件
###收集logstash自身的日志，172.20.22.26
# cd /etc/logstash/conf.d
# cat java.conf
  |4 _! u& d+ s- a+ G$ ainput {
$ U& y( H9 H" s6 S8 n3 h& h0 v  file {
7 B. Y( `+ ^1 d2 K+ X: j1 x    path => "/var/log/logstash/logstash-plain.log"
/ Q8 u: [' y* u0 _7 K) \    type => "logstash-log"
    start_position => "beginning"
( o1 {7 S- ^3 ]# |6 `: x- r    stat_interval => "3"
$ b8 D+ O+ [9 ]4 y' d% P% O8 j    codec => multiline {
; r% O3 F7 H" n' Q  y2 _- z      pattern => "^\["
( x% U+ d& w' @4 Y8 m      negate => true
      what => "previous"
0 ^/ s$ b; w: L! i* v9 X! `   }}
% O( b% h& g  n" E) m}
: k2 I5 X/ j2 K! D2 k) boutput {
  if [type] == "logstash-log" {
! |. ]( `$ X/ e  elasticsearch {
4 v' g6 J# v' H    hosts => ["172.20.22.24"]
8 u/ a) C2 E0 h2 F: [. u( U    index => "logstash-log-%{+YYYY.MM.dd}"
  }}
8 x5 K9 y) u# d}
4 [- z8 {, i& h- Q! V! q" h( O
+ z; Y5 m6 B5 J% W) m- n0 y* ?4 H3 R- w# /usr/share/logstash/bin/logstash -f java.conf -t
1 L4 P0 x: C% _( T6 f' l- E$ a# systemctl restart logstash.service

$ ~* ]/ y& }, H! |/ j###收集logstash自身的日志，172.20.22.30
# cd /etc/logstash/conf.d
( B( ~$ @- }( W: Q0 ~' S0 I6 A# cat java.conf
2 e# ]% U9 j; n5 tinput {
0 V$ C- P' h, Z1 m3 C% s4 F- E1 D! Q  file {
    path => "/var/log/logstash/logstash-plain.log"
2 z, b, r8 u. A1 x5 O# @9 _    type => "logstash-log"
* X  C) O2 C: S# U    start_position => "beginning"
6 }) N) T) m6 U  D/ y) z5 v    stat_interval => "3"
; A$ Y6 q& c8 w% Y) d    codec => multiline {
      pattern => "^\["
      negate => true
      what => "previous"
' F# x0 a# {) l5 n& s! L   }}
}
output {
7 i7 {8 m: r/ ]+ K$ ~7 m- R0 G  if [type] == "logstash-log" {
  elasticsearch {
0 \5 V6 D; p$ F6 c    hosts => ["172.20.22.24"]
* e5 j  B% |! G. M* L    index => "logstash-log-%{+YYYY.MM.dd}"
" y% d# Q1 A5 W' `6 d  }}
}
& S0 J$ n0 y1 q; o0 T' `$ T
* l' i1 j1 ]' L1 Q( z- b! z# /usr/share/logstash/bin/logstash -f java.conf -t
% }; C2 X) ]% }8 K; r4 A9 `& K# systemctl restart logstash.service
" A* M$ }6 m, u# a0 v5 l查看kibana收集到的日志

: [5 I$ [. Y( n! A4 a! R% \
% C6 y+ ?8 f1 _2 W; o( u1 a8 z
filebeat结合redis、logstash收集nginx日志
) L1 @- r( S3 I! ?使用filebeat收集日志发送到logstash1，再由logstash1发送到redis，最后再由logstash2发送到elasticsearch

web1：172.20.22.30，部署好nginx、filebeat、llogstash

web2：172.20.22.26，部署好nginx、filebeat、llogstash

logstash服务器2：172.20.22.23，redis服务器：172.20.23.157

5 M! u  l  ~, J) ^1 I* mnginx服务器相关配置
部署nginx
. o" V* |* N$ U6 m; |2 Q# wget http://nginx.org/download/nginx-1.18.0.tar.gz
2 g9 |5 I3 L$ W1 d! W# tar xf nginx-1.18.0.tar.gz
) L5 d4 p" ?/ E. p' R# cd nginx-1.18.0
9 B+ ^* `5 t- F- b0 q  Q* n# ./configure --prefix=/usr/local/nginx --with-http_ssl_module
# make -j4 && make install
# /usr/local/nginx/sbin/nginx
& g% a: q0 R6 ^/ G0 u+ h部署配置logstash
把filebeat收集到的日志信息发送到redis
* ?$ s9 R5 d, C+ U8 q
! U- _9 z4 K2 U8 P) [# apt install -y openjdk-8-jdk
# dpkg -i logstash-7.12.1-amd64.deb
# cat /etc/logstash/conf.d/beats-to-redis.conf
. Y) X1 m( }" M( ^input {
: f" m/ e) N* r  beats {
    port => 5044
    codec => "json"
  }
  beats {
7 Q9 a) e$ n( f    port => 5045
    codec => "json"
  }
}
output {
  if [fields][project] == "filebeat-systemlog" {
    redis {
      data_type => "list"
      key => "filebeat-redis-systemlog"
      host => "172.20.23.157"
      port => "6379"
      db => "0"
      password => "12345678"
  }}
* B" W4 m) g. p2 B0 M  if [fields][project] == "filebeat-nginx-accesslog" {
! R* }( ?9 H3 w% L2 J) c    redis {
      data_type => "list"
7 \1 b$ P& e+ F# n% N4 k      key => "filebeat-redis-nginx-accesslog"
) ^: M. b0 ]+ d; T, X      host => "172.20.23.157"
      port => "6379"
      db => "1"
      password => "12345678"
  }}
* ~4 V, b3 L1 S  if [fields][project] == "filebeat-nginx-errorlog" {
    redis {
      data_type => "list"
      key => "filebeat-redis-nginx-errorlog"
      host => "172.20.23.157"
& f) h* D8 }0 V0 U# e      port => "6379"
! l! F  a: |/ ~0 a& t0 ^2 _      db => "1"
3 m/ C2 f/ Z4 b      password => "12345678"
! r3 U6 B. R* p* H/ r! J# V  }}
}
# systemctl start logstash
# scp /etc/logstash/conf.d/beats-to-redis.conf root@172.20.22.26:/etc/logstash/conf.d/
部署配置filebeat
  L2 a, u' Z7 e+ M/ H通过filebeat收集日志信息发送到logstash
; s$ D3 i1 I+ T  i2 c
4 m8 |( e! R& ?; o* d9 V# dpkg -i filebeat-7.12.1-amd64.deb
# grep -v "#" /etc/filebeat/filebeat.yml | grep "^[^$]"
" N$ h, e/ A1 T6 d4 h" r, vfilebeat.inputs:
0 G& B* D0 v, [9 c5 A' U- type: log
  enabled: true
  paths:
    - /var/log/syslog
$ B0 }4 @( P- f& v  fields:
  u: ~. H3 l) _& ~( D2 c' }# f    project: filebeat-systemlog
- type: log
  enabled: true
  paths:
- q, b( i& `# u    - /usr/local/nginx/logs/access.log
8 B( S9 T& y: Y  fields:
* U2 N$ G- ~! }9 }" o& `0 Q    project: filebeat-nginx-accesslog
- type: log
$ T" X# P2 {* g7 ^7 y3 L- T/ x' f/ a  enabled: true
( X# y9 }% k. R( k/ l  paths:
    - /usr/local/nginx/logs/error.log
3 P, h2 S6 x4 l) j  fields:
! R- i" b8 O  l# w" M: u6 n- Q    project: filebeat-nginx-errorlog
1 ^( ?$ ~( i1 R2 U- ?filebeat.config.modules:
8 X/ [  J) n5 ]; z( }4 k! i  path: ${path.config}/modules.d/*.yml
  reload.enabled: false
: W" I# |. }; i+ hsetup.template.settings:
  index.number_of_shards: 1
- C) h" C5 A  L/ p' M. ssetup.kibana:
% W. r( _. Z" O; a2 ^8 D2 Hprocessors:
! [! r+ [8 h: E; j  - add_host_metadata:
  a; J* f- n) f3 `/ R$ X& C      when.not.contains.tags: forwarded
  - add_cloud_metadata: ~
  - add_docker_metadata: ~
7 p7 P8 Q0 d2 T( Z$ j  - add_kubernetes_metadata: ~
output.logstash:
8 j" t  Z& }9 T) O* ]3 C0 M" R  hosts: ["172.20.22.30:5044","172.20.22.30:5045"]
" {; u/ G/ n2 M! Q* ?- l6 [, t& q  enabled: true
$ g8 {  [6 {% I/ u  worker: 2
  compression_level: 3
  loadbalance: true

; e* m( f! ^; t2 h$ _# systemctl start filebeat
# scp /etc/filebeat/filebeat.yml root@172.20.22.26:/etc/filebeat/
logstash服务器配置
0 _& r2 L4 r7 ^1 M$ x' Alogstash服务器2：172.20.22.23，把redis缓存的日志发送到elasticsearch

# apt install -y openjdk-8-jdk
/ q1 I3 ?% `1 K3 _3 `# dpkg -i logstash-7.12.1-amd64.deb
# cat /etc/logstash/conf.d/redis-to-es.conf
input {
+ ?: l7 f" n, z+ G. K  redis {
* z* x" z# L8 x7 n+ T6 C+ b! `: T    data_type => "list"
    key => "filebeat-redis-nginx-accesslog"
    host => "172.20.23.157"
" u; I( A! L& ~4 d& H    port => "6379"
+ b6 f/ q9 E2 a- D8 a+ }( t2 |, z; n    db => "1"
    password => "12345678"
1 `( l. M# r2 u1 h7 o  }
  redis {
    data_type => "list"
( ~# `& c6 Z$ B/ R    key => "filebeat-redis-nginx-errorlog"
& E2 V: c3 e3 X0 }( X$ B    host => "172.20.23.157"
0 v+ ?# f- [2 L1 O& n1 I* [! O    port => "6379"
2 J5 M& f! H: q1 @4 E# T" O    db => "1"
    password => "12345678"
  }
$ z& V8 y7 [/ t, }5 r1 J- V; P  redis {
    data_type => "list"
    key => "filebeat-redis-systemlog"
    host => "172.20.23.157"
+ s# }" ~. H& t$ u$ q1 }% N    port => "6379"
    db => "0"
1 n4 @: a6 p; n* s! |# ?4 S    password => "12345678"
  }
}
output {
  if [fields][project] == "filebeat-systemlog" {
% k" I, K% C; V    elasticsearch {
1 g) \, P. u" W      hosts => ["172.20.22.28:9200"]
  e1 ?* t+ S0 P) U+ z0 S      index => "filebeat-systemlog-%{+YYYY.MM.dd}"
& M( ]% P' J" I( [. o  }}
3 s* w8 w3 s8 \! O  if [fields][project] == "filebeat-nginx-accesslog" {
    elasticsearch {
# E5 c$ ^- Q: d' U      hosts => ["172.20.22.28:9200"]
" T& V, g* R+ P  l" a      index => "filebeat-nginx-accesslog-%{+YYYY.MM.dd}"
  }}
, v2 q& w7 Q' _+ T  if [fields][project] == "filebeat-nginx-errorlog" {
    elasticsearch {
9 g7 C; q  j1 D; f+ F( J, Y, s% f      hosts => ["172.20.22.28:9200"]
      index => "filebeat-nginx-errorlog-%{+YYYY.MM.dd}"
  }}
}
0 _) A* y  t  f# systemctl restart logstash.service
- b! k/ a9 `, h0 t3 e4 G3 T, Zredis安装配置
redis服务器：172.20.23.157，
6 S$ H* K6 D$ \  `! C+ ^0 V' _
# yum install -y redis
# vim /etc/redis.conf
3 u5 ^1 [  [; g- {4 b& |####修改以下配置项
. Q- J; c6 f8 N, xbind 0.0.0.0
; I" m( N# d* W( o" B6 y....
5 }# D3 x* [6 k; u  F' s# csave ""
....
7 B7 _6 L+ f  Y  W' l9 d9 srequirepass 12345678
....
# I6 a) _" n: t1 T0 M2 G3 b# systemctl start redis
7 C: E  @. e. [0 I###测试连接redis
; Z. |: C/ i  |/ T# redis-cli
0 [* O3 X6 z9 E127.0.0.1:6379> auth 12345678
OK
, s2 j. J! ^# q3 ~: \$ ~; i. y127.0.0.1:6379> ping
PONG

  Q! k% O, k, F- p* R###验证收集到的日志信息
127.0.0.1:6379[1]> keys *
) b9 n, B# u" d( `4 ?1) "filebeat-redis-nginx-accesslog"
; m; T  j  Z7 v7 g2 M1 \) E* F( Z2) "filebeat-redis-nginx-errorlog"
+ r2 P+ E* n/ s: J+ Z! Q" @127.0.0.1:6379[1]> select 0
OK
3 M9 x$ i( M/ X6 p# {+ H2 ~127.0.0.1:6379> keys *
; A; O, b" F* ~; A6 y1) "filebeat-redis-systemlog"
通过head插件验证生成的索引


kibana验证收集到的日志信息