Logstash完成ELK集群

注：本文与同步(9条消息) 搭建Elasticsearch和kibana完成初步ELK日志分析平台搭建_kali_yao的博客-CSDN博客

logstash搭建

1)logstash介绍

什么是logstash

-是一个数据采集、加工处理以及传输的工具

特点

-所有类型的数据集中处理

-不同模式和格式数据的正常化

-自定义日志格式的迅速扩展

-为自定义数据源轻松添加插件

Logstash工作结构(与流水线类似)

{数据源} ==> input{} ==> filter{ } ==> output {} ==>{输出结果}

- input 负责收集数据

-filter负责处理数据

-output负责输出数据

logstash 里面的类型

-布尔值类型: ssI_ enable => true

-字节类型: bytes =>1MiB"

-字符串类型: name => " xkops'

-数值类型: port => 22

-数组: match => [" datetime", "UNIX"]

-哈希(键值对) : options => {k => "v",k2 => "v2"}

-注释: #

logstash 里面的判断语法

-等于: ==

-不等于: !=

-小于: <

-大于: >

-小于等于: <=

-大于等于: >=

-匹配正则:-不匹配正则: !

logstash 里面的逻辑判断

-包含: in

-不包含: notin

-与: and

-或: or

-非与: nand

-非或: xor

其他

-编码类型: codec => "json'

2)logstash安装

logstash安装概述

-ogstash依赖Java环境，需要安装java-openjdk

-logstash没有默认的配置文件，需要手动配置

-logstash安装在/usr/share/logstash 目录下

配置好yum源以后直接使用yum进行安装或手动下载logstash
~]# wget https://artifacts.elastic.co/downloads/logstash/logstash-6.8.8.rpm
~]# yum install -y java-1.8.0-openjdk logstash

logstash安装准备

主机配置

主机	IP地址	配置
logstash	192.168.1.47	最低配置2核2G

logstash基础配置安装

# 注需于前面的配置elasticsearch对应
~]# vim /etc/hosts
192.168.1.41    es-0001
192.168.1.42    es-0002
192.168.1.43    es-0003
192.168.1.44    es-0004
192.168.1.45    es-0005
192.168.1.47    logstash
~]# yum install -y java-1.8.0-openjdk logstash
# 默认安装在/usr/share/logstash/bin/logstash
~]# cd /usr/share/logstash/bin/ && ls
注： /usr/share/logstash/bin/logstash直接执行会报错没找到配置文件
# 服务配置文件默认一般在/etc下
~]# ln -s /etc/logstash /usr/share/logstash/config
~]# vim /etc/logstash/conf.d/my.conf
input {stdin {}
}

filter{ }

output{stdout{}
}
~]# /usr/share/logstash/bin/logstash
做完软连接之后直接启动会告诉我没没用配置文件，（配置文件需要自己书写，如上）

注：启动后进入交互视，输入任意测试，回应json格式

ctrl + c退出

3)插件的使用

logstash 配置文件路径

-logstash安装路径/usr/share/ logstash (默认路径)

-配置文件安装路径/etc/logstash

●启动命令logstash

/usr/share/ logstash/bin/logstash

插件的管理

使用插件

-面的配置文件使用了logstash-input-stdin 和

logstash-output- -stdout两个插件，logstash对数据的处理依赖插件

管理命令/usr/share/ logstash/bin/ logstash-plugin

-安装插件install

-删除插件uninstall

-查看插件list
~]# cd /usr/share/logstash/bin
~]# ./logstash-plugin list
# 注：同一种插件只能工作在同一个工作区里
只能用于input段的插件: Iogstash-input- -xxx

只能用于output段的插件: logstash-output- -xxx

只能用于filter段的插件: logstash-filter-xxx,

编码格式插件: logstash-codec-xxx

插件与调试格式

/usr/share/logstash/bin/logstash-plugin list #查看所有

使用json格式字符串测试 {"a":"1", "b":"2", "c":"3"}

# 注：当没有输入格式之前所有的输入都会打印成普通字符串
~]# vim /etc/logstash/conf.d/my.conf
input { stdin { codec => "json" }  # 以json格式打印
}

filter{ }

output{ stdout{ codec => "rubydebug" } #调试格式
}
~]# /usr/share/logstash/bin/logstash
[2021-09-09T10:28:49,544][INFO ][logstash.agent           ] Successfully started Logstash API endpoint {:port=>9600}{"a":"1", "b":"2", "c":"3"} # 输入
/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/awesome_print-1.7.0/lib/awesome_print/formatters/base_formatter.rb:31: warning: constant ::Fixnum is deprecated
{"@version" => "1","c" => "3","host" => "logstash","@timestamp" => 2021-09-09T02:28:52.117Z,"b" => "2","a" => "1"
}

服务启动报错（由于上一个服务关闭进程没有关闭则需要关闭上一个进程）

此时普通的查进程以经不能解决

~]# ps -exf | grep logstash
# 因为logstash是java启动的，这边安装一个java的命令工具查找并杀死
~]# yum install java-1.8.0-openjdk-devel
~]# jps
629 WrapperSimpleApp
2078 Jps
~]# kill 2078

4)插件官方手策使用

官方手册地址

Logstash Reference [8.0] | Elastic

注：这里我用的是6.8

下面三个模式任选一个我这里演示input

这里演示file文件

找到path这一行点array有案例

案例可以直接使用

查看使用规则

5)input file插件（文件插件）

~]# vim /etc/logstash/conf.d/my.conf
input { stdin { codec => "json" }file {path => ["/tmp/b.log","/tmp/c.log"]        # 指定监控的文件当有多条的时候加[]type => "test"   # 指定标签，如果是apache的日志可以打上apachestart_position => "beginning" # 开始,如果不写的话在启动服务之前的数据不会打印      sincedb_path => "/var/lib/logstash/sincedb"  # 书签文件}file {              # 或者可以支持多个file书写path => "/tmp/d.log"type => "mysql"}
}

}
filter{ }
output{ stdout{ codec => "rubydebug" }
}
~]# rm -rf /var/lib/logstash/plugins/inputs/file/.sincedb_*
# 删除原有书签文件，这样就可以重头开始输出了
~]# /usr/share/logstash/bin/logstash

写入数据测试

~]# cd /tmp
~]# echo aaa >>a.log
~]# echo bbb >>a.log
~]# /usr/share/logstash/bin/logstash
....
/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/awesome_print-1.7.0/lib/awesome_print/formatters/base_formatter.rb:31: warning: constant ::Fixnum is deprecated
{"message" => "aaa","@version" => "1","@timestamp" => 2021-09-09T03:57:10.178Z,"host" => "logstash","type" => "test","path" => "/tmp/c.log"
}
{"message" => "bbb","@version" => "1","@timestamp" => 2021-09-09T03:57:10.200Z,"host" => "logstash","type" => "test","path" => "/tmp/c.log"

官网查看解析

6)web页面展示

# 浏览器访问apache服务的ip（之前有写）
~]# http://172.17.0.51/info.php

查看日志

~]# tail -f /etc/httpd/logs/access_log
172.17.0.126 - - [07/Mar/2022:11:42:03 +0800] "GET /info.php HTTP/1.1" 200 279 "-" "Mozilla/5.0 (Windows NT 6.3; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/85.0.4183.83 Safari/537.36"

# 第一个字段是ip，发现不是本机ip访问的

默认在云上访问使用的是LB负载ip

设置监听服务器时打开获取真实ip

测试，把apache的日志拷贝到/tmp/b.log中，启动查看

此时发现只要再次启动就不会打印输出了，因为有标签，

修改标签指定文件或删除标签文件

7)filter（过滤） grok插件

1）filter插件配置管理的概述

grok插件

-解析各种非结构化的日志数据插件

-grok使用正则表达式把非结构化的数据结构化在分组匹配-正则表达式需要根据具体数据结构编写

-虽然编写困难，但适用性极广

-几乎可以应用于各类数据
filter{ grok {match => { "message" => "" }}
}

2）filter插件配置的正则概述

正则表达式分组匹配格式:

调用格式：(?<名字>正则表达式)

## 以apache日志为例写一个正则
~]# vim /etc/logstash/conf.d/my.conf

input {stdin { codec => "json" }file {path => ["/tmp/b.log","/tmp/c.log"]type => "test"start_position => "beginning"sincedb_path => "/dev/null"}
}

filter{grok {match => { "message" => "(?<client_IP>(25[0-5]|2[0-4]\d|1?\d?\d\.){3}(25[0-5]|2[0-4]\d|1?\d?\d)) (?<vber>\S+) (?<auth>\S+) \[(?<time>.+)\] \"(?<method>[A-Z]+) (?<url>\S+) (?<proto>[A-Z]+\/\d\.\d)\" (?<rc>\d+) (?<size>\d+) \"(?<ref>\S+)\" \"(?<agent>[^\"]+)\"" }}
}

output{stdout{ codec => "rubydebug" }
}

正则表达式宏调用格式

调用格式: %{宏名称:名字}

~]# vim /etc/logstash/conf.d/my.conf
input {stdin { codec => "json" }file {path => ["/tmp/b.log","/tmp/c.log"]type => "test"start_position => "beginning"sincedb_path => "/dev/null"}
}

filter{grok {match => { "message" => "%{IP:client_IP}" }   # 用宏}
}

output{stdout{ codec => "rubydebug" }

宏文件路径

/usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-patterns-core-4.1.2/patterns

]# cd /usr/share/logstash/vendor/bundle/jruby/2.5.0/gems/logstash-patterns-core-4.1.2/patterns
~]# ls
aws     exim           httpd         maven                 nagios      ruby
bacula  firewalls      java          mcollective           postgresql  squid
bind    grok-patterns  junos         mcollective-patterns  rails
bro     haproxy        linux-syslog  mongodb               redis

查看IP的宏

## 用httpd宏示例
~]# echo '192.168.1.252 - - [29/Jul/2020:14:06:57 +0800] "GET /info.html HTTP/1.1" 200 119 "-" "curl/7.29.0"' >/tmp/c.log
~]# vim /etc/logstash/conf.d/my.conf
input {file {path => ["/tmp/c.log"]type => "test"start_position => "beginning"sincedb_path => "/dev/null"     #把日志位置指针指向空可以反复读取}
}
filter{grok {match => { "message" => "%{HTTPD_COMBINEDLOG}" } # 调用httpd的宏把匹配出日志的各个字段含义转化成json格式}
}
output{stdout{ codec => "rubydebug" }
}
~]# /usr/share/logstash/bin/logstash

8)output elasticsearch插件

~]# vim /etc/logstash/conf.d/my.conf
input {file {path => ["/tmp/c.log"]type => "test"start_position => "beginning"sincedb_path => "/dev/null"}
}
filter{grok {match => { "message" => "%{HTTPD_COMBINEDLOG}" }}
}
output{stdout{ codec => "rubydebug" }elasticsearch {hosts => ["es-0004:9200", "es-0005:9200"]  # 把数据写入集群（注，为了防止单点故障，head（es-0001）,kibana(es-002,es-003),所以这里就用4和5）index => "weblog-%{+YYYY.MM.dd}"}   # 索引的名字用来存放日志
}
~]# /usr/share/logstash/bin/logstash

# 浏览器打开 head 插件，通过 web 页面浏览验证http://公网IP:9200/_plugin/head/

查看内容

8.远程获取WEB日志

1)beats插件

如何收集日志?

-由于logstash 依赖JAVA环境，而且占用资源非常大，因此在每一台web 服务器上部署logstash 非常不合适

-我们可以使用更轻量的filebeat 收集日志，通过网络给logstash发送数据

- logstash 使用beats 接收日志，完成分析

logstash接收日志

-如果想接收数据，必须监听网络服务。logstash 可以通过beats插件接收filebeats 发送过来的数据

2)logstash beats插件配置

~]# vim /etc/logstash/conf.d/my.conf
input { stdin { codec => "json" }file{path => ["/tmp/c.log"]type => "test"start_position => "beginning"sincedb_path => "/var/lib/logstash/sincedb"}beats {          # 配置beatsport => 5044}
}

filter{ grok {match => { "message" => "%{HTTPD_COMBINEDLOG}" }}
}

output{ stdout{ codec => "rubydebug" }elasticsearch {hosts => ["es-0004:9200", "es-0005:9200"]index => "weblog-%{+YYYY.MM.dd}"}
}
~]# /usr/share/logstash/bin/logstash

# 测试，开启另一个终端查看
~]# ss -nulpt | grep 5044

3)web服务安装filebeat

下载软件包

或者

~]# wget https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-6.8.8-x86_64.rpm

# 在web服务器上操作
~]# yum install -y filebeat.x86_64 0:6.8.8-1
~]# vim /etc/filebeat/filebeat.yml
24:  enabled: true    # 打开模块
28:  - /var/log/httpd/access_log  # 监控日志文件名
45:    fields:        # 定义标签
46:       my_type: apache # 改成自己的标签
148, 150 注释掉            # 因为日志不是json格式不能直接写
161: output.logstash:
163:   hosts: ["(logstash的ip):5044"]
180, 181, 182 注释掉  # 收集系统性息

# 查看修改
~]# grep -Pv "^\s*(#|$)" /etc/filebeat/filebeat.yml
# 启动filebeat服务（最好重启一下apache服务，为了更好的识别filebeat）
~]# systemctl enable --now filebeat

4.验证

配置完成以后可以使用grep查看
~]# grep -Pv "^\s*(#|$)" /etc/filebeat/filebeat.yml
# 启动服务
~]# /usr/share/logstash/bin/logstash
访问web服务器查看

kibana配置

排错思路

# 1.查看端口是否启动（5044）
~]# ss -nulpt | grep 5044
# 2.查看配置文件是否有问题
~]# cat /etc/logstash/conf.d/my.conf
# ping一下指定的elasticsearch集群的ip
# 3.查看elasticsearch数据库集群的状态（red为错误状态）
~]# curl http://node-2:9200/_cat/indices
# 4.查看集群(如果没有启动则需要到所有服务上启动elasticsearch)
~]# curl http://node-2:9200/_cat/nodes
# 查看是否恢复
~]# curl http://node-2:9200/_cluster/health?pretty

访问测试

 ~]# ab -c 20 -n 200 http://172.17.0.51/info.php-c # 并发-n # 次数

直接选择查看ip

再次拆分（可以看到是apache的ab压力测试）